RabbitMq 操作总结

时间 2021-01-11

RabbitMq java操作总结

消息队列解决了什么问题?

异步处理
应用解耦
流量削锋
日志处理

JAVA操作rabbitmq

Simple 简单队列
work queues工作队列公平分发轮询分发
publish/subscribe发布订阅
Routing 路由选择通配符模式
Topics 主题
手动和自动确认消息
队列的持久化和非持久化

RabbitMq安装步骤：

https://blog.csdn.net/qq_41950069/article/details/81346675

http://www.noobyard.com/article/p-kpczytby-go.html

先安装erlong
再安装rabbitmq

Rabbit mq 教程：

https://blog.csdn.net/hellozpc/article/details/81436980

安装好后rabbitmq后台登录地址：http://localhost:15672/

账号密码：都是guest

Spring boot整合RabbitMq:

http://www.noobyard.com/article/p-zbdqsmbb-et.html

http://www.noobyard.com/article/p-tjutigmg-hy.html

SpringBoot整合RabbitMQ之典型应用场景实战一:

http://www.noobyard.com/article/p-rrnndfmc-dp.html

springboot+rabbitMq整合开发实战二：模拟用户下单的过程:

http://www.noobyard.com/article/p-vnouhizv-b.html

1、添加用户

用户界面

Virtual hosts 管理

Virtual hosts 相当于mysql 的 db ，添加的路径一般以“/”开头

添加路径后在上面的表格中就会显示你添加的路径，点击添加的路径就会进入到下面的界面

给用户进行授权，选择添加的用户进行授权，这样用户才能访问这个地址

2、java操作队列

2.1简单队列

2.1.1 模型

p：消息的生产者

红色区域：队列（Queue）

C：消费者

3个对象：生产者队列消费者

RabbitMQ中的消息都只能存储在Queue中，生产者（图中的P）生产消息并最终投递到Queue中，消费者（图中的C）可以从Queue中获取消息并消费。

2.1.2 获取MQ连接

Java代码演示：

1、引入maven依赖

<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.3</version>
</dependency>

2、编写连接工具类

public class RabbitConnectionUtil {
//在rabbitmq页面自己创建的VirtualHost地址
public final static String vHost = "/test_rabbit";
/**
* 创建连接方法
* @return 连接
*/
public static Connection getConnection() throws Exception {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//设置服务器地址
connectionFactory.setHost("127.0.0.1");
//设置AHQP 连接端口
connectionFactory.setPort(5672);
//设置VirtualHost 地址
connectionFactory.setVirtualHost(vHost);
//设置用户名
connectionFactory.setUsername("guest");
//设置密码
connectionFactory.setPassword("guest");
return connectionFactory.newConnection();
}
}

2.1.3 生产者生产消息

1、首先需要定义队列的名称，这里我将队列名称定义提取出来作为常量，生产者和消费者都是在这个队列中发送和接收消息

public class RabbitConstant {
public final static String QUEUE_NAME = "test_simple_queue";
}

2、发送队列消息

public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME,false,false,false,null);
//向队列中推送消息
String msg = "hello simple !";
channel.basicPublish("",RabbitConstant.QUEUE_NAME,null,msg.getBytes());
System.out.println("send simple success："+msg);
//关闭通道和连接
channel.close();
connection.close();
}

2.1.4 消费者接收消息

1、老版本支持的方法，亲测3.x之类的版本支持这种方式，在5.版本已经不支持这种方式

public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列消费者
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
//监听队列
channel.basicConsume(RabbitConstant.QUEUE_NAME,true,queueingConsumer);
while (true){
Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("Recv success:"+msg);
}
}

2、新版本支持方式

public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME,false,false,false,null);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("Recv success："+msg);
}
};
//监听队列
channel.basicConsume(RabbitConstant.QUEUE_NAME,true,consumer);
}

2.1.5 简单队列的不足

耦合性高,生产者--对应消费者(如果我想有多个消费者去消费队列中消息,这时候就不行了)队列名称变更这时候得同时变更

2.2 工作队列（轮询分发 round-robin）

2.2.1 模型

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

为什么会出现工作队列？

Simple队列是一一对应的,而且我们实际开发,生产者发送消息是毫不费力的,而消费者一般是要跟业务相结合的，消费者接收到消息之后就需要处理，处理需要花费时间,这时候队列就会积压了很多消息。

2.2.2 生产者

1、首先需要定义队列的名称，这里我将队列名称定义提取出来作为常量，生产者和消费者都是在这个队列中发送和接收消息

public class RabbitConstant {
public final static String QUEUE_NAME_SIMPLE = "test_simple_queue";
public final static String QUEUE_NAME_WORK = "test_work_queue";
}

2、创建生产者

public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_WORK,false,false,false,null);
//向队列中推送消息
for(int i = 0; i<50;i++){
String msg = "hello work["+i+"] !";
channel.basicPublish("",RabbitConstant.QUEUE_NAME_WORK,null,msg.getBytes());
System.out.println("send work success："+msg);
Thread.sleep(20);
}
//关闭通道和连接
channel.close();
connection.close();
}

2.2.3 消费者

消费者1：

public class Recv1 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE,false,false,false,null);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【1】Recv success："+msg);
try {
Thread.sleep(600);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【1】 done ！");
}
}
};
//监听队列
Boolean autoAck = true;
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE,autoAck,consumer);
}
}

消费者2：

public class Recv2 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE,false,false,false,null);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【2】Recv success："+msg);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【2】 done ！");
}
}
};
//监听队列
Boolean autoAck = true;
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE,autoAck,consumer);
}
}

2.2.4 现象

消费者1休眠时间在600ms和消费者2休眠时间在200ms的情况下，消费者1和消费者2处理的消息是一样多的
消费者1:偶数
消費者2:奇数
这种方式叫做轮询分发(round-robin)结果就是不管谁忙活谁清闲，都不会多给一个消息，任务总是你一个我一个

2.3 工作队列（公平分发fail dipatch）

2.3.1 生产者

public class send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_WORK,false,false,false,null);
//在每个消费者发送确认消息之前，消息队列不发送下一个消息到消费者，1次只处理1个消息
//限制发送给同一个消费者不得超过一次
int prefethCount = 1;
channel.basicQos(prefethCount);
//向队列中推送消息
for(int i = 0; i<50;i++){
String msg = "hello work["+i+"] !";
channel.basicPublish("",RabbitConstant.QUEUE_NAME_WORK,null,msg.getBytes());
System.out.println("send work success："+msg);
Thread.sleep(20);
}
//关闭通道和连接
channel.close();
connection.close();
}
}

2.3.2 消费者

消费者1：

public class Recv1 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE,false,false,false,null);
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【1】Recv success："+msg);
try {
Thread.sleep(600);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【1】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE,autoAck,consumer);
}
}

消费者2：

public class Recv2 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE,false,false,false,null);
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【2】Recv success："+msg);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【2】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE,autoAck,consumer);
}
}

2.3.3 现象

消费者2处理的消息比消费者1多，能者多劳。因为在消费者1中设置的是每次处理完一个消息休眠600ms，而消费者2是休眠200毫秒，所以消费者2比消费者1多。

2.4 消息应答与消息持久化

2.4.1 消息应答

//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE,autoAck,consumer);

Boolean autoAck = true;(自动确认模式)，一旦rabbitmq将消息分发给消费者，就会从内存中删除

这种情况下，如果杀死正在执行的消费者，就会丢失正在处理的消息

Boolean autoAck = false;(手动模式)，如果有1个消费者挂掉，就会交付给其他消费者，rabbitmq支持消息应答,消费者发送一个消息应答,告诉rabbitmq这个消息我已经处理完成你可以删了,然后rabbitmq就会删除内存中的消息

消息应答默认是打开的，false

Message acknowledgment:(消息应答)

大家想，如果我们的rabbitmq挂了，我们的消息依然会丢失

2.4.2 消息的持久化

//创建队列声明
boolean durale = false;
channel.queueDeclare(RabbitConstant.QUEUE_NAME_WORK,durale,false,false,null);

durale:表示是否持久化，false表示不进行持久化，true表示持久化

我们将程序中的boolean durale = false;改成true是不可以的，尽管代码是正确的，它也不会执行成功，因为我们已经定义了一个叫做“test_work_queue”的队列，这个queue (队列) 是未持久化的，rabbitmq不允许重新定义（不同参数）一个已经存在的队列。如果需要重新定义这个队列的参数，唯一的做法就是进入rabbitmq将这个队列删除，再重新启动java程序，这时就可以重新创建一个修改后参数的队列。

2.4.3 queueDeclare方法（队列声明）

Queue.DeclareOk queueDeclare (String queue , boolean durable , boolean exclusive , boolean autoDelete , Map arguments) throws IOException;

该方法有5各参数：

queue 队列的名称

durable: 设置是否持久化。为 true 则设置队列为持久化。持久化的队列会存盘，在服务器重启的时候可以保证不丢失相关信息。

exclusive 设置是否排他。为 true 则设置队列为排他的。如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点:排他队列是基于连接( Connection) 可见的，同个连接的不同信道 (Channel) 是可以同时访问同一连接创建的排他队列; "首次"是指如果个连接己经声明了排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同:即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景。

autoDelete: 设置是否自动删除。为 true 则设置队列为自动删除。自动删除的前提是: 至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。不能把这个参数错误地理解为: "当连接到此队列的所有客户端断开时，这个队列自动删除"，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列。

argurnents: 设置队列的其他一些参数，如 x-rnessage-ttl 、x-expires 、x-rnax-length 、x-rnax-length-bytes、 x-dead-letter-exchange、 x-deadletter-routing-key 、 x-rnax-priorit

2.5 订阅模式（publish_subscribe）

2.5.1 模型

解读：

1,一个生产者,多个消费者

2,每一个消费者都有自己的队列

3.生产者没有直接把消息发送到队列而是发到了交换机转发器exchange

4,每个对列都要绑定到交换机上

5.生产者发送的消息经过交换机到达队列，就能实现一个消息被多个消费者消费

2.5.2 生产者

首先定义交换机名称

public class RabbitConstant {
public final static String QUEUE_NAME_SIMPLE = "test_simple_queue";
public final static String QUEUE_NAME_WORK = "test_work_queue";
public final static String EXCHANGE_NAME_FANOUT = "test_fanout_exchange";//交换机名称
}

2.创建生产者

public class send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(RabbitConstant.EXCHANGE_NAME_FANOUT,"fanout");//分发
//向交换机中推送消息
String msg = "hello publish subscribe !";
channel.basicPublish(RabbitConstant.EXCHANGE_NAME_FANOUT,"", null, msg.getBytes());
System.out.println("send publish subscribe success：" + msg);
//关闭通道和连接
channel.close();
connection.close();
}
}

通过控制台可以看到，并没有我们发送的消息，消息哪去了?

丢失了，因为交换机没有存储的能力,在rabbitmq里面只有队列有存储能力，因为这时候还没有队列绑定到这个交换机所以数据丢失了。

2.5.3 消费者

首先定义绑定到交换机的队列的名称

public class RabbitConstant {
public final static String QUEUE_NAME_SIMPLE = "test_simple_queue";
public final static String QUEUE_NAME_WORK = "test_work_queue";
public final static String EXCHANGE_NAME_FANOUT = "test_fanout_exchange";//交换机名称
public final static String QUEUE_NAME_EXCHANGE_BIND_EMIL = "test_queue_fanout_emil";//绑定到交换机的队列
public final static String QUEUE_NAME_EXCHANGE_BIND_SMS = "test_queue_fanout_sms";//绑定到交换机的队列
}

2、创建消费者

消费者1：

public class Recv1 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_EMIL,false,false,false,null);
//绑定队列到交换机
channel.queueBind(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_EMIL,RabbitConstant.EXCHANGE_NAME_FANOUT,"");
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【1】Recv success："+msg);
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【1】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_EMIL,autoAck,consumer);
}
}

消费者2：

public class Recv2 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_SMS,false,false,false,null);
//绑定队列到交换机
channel.queueBind(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_SMS,RabbitConstant.EXCHANGE_NAME_FANOUT,"");
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【2】Recv success："+msg);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【2】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_SMS,autoAck,consumer);
}
}

2.5.4 总结

在控制界面可以看到有两个队列已经绑定到了转换器上

测试结果：同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。这个怎么说呢，在同一个队列中呢，只能有一个消费者可以获得消息，因为这是点对点，是队列的特性，谁把这个机会抢走了，这个机会就没有了。

发布订阅类似广播，发送方只管向外发送消息，接收方呢只要去听这个广播就能得到广播的消息，但是如果你刚好不在，那么广播里的内容也就错过了。在rabbitmq中就是绑定到交换机的队列的消费者可以接收到这个消息。

2.6 exchange (交换机转换器)

2.6.1 说明

作用：一方面是接收生产者的消息,另一方面是向队列推送消息

2.6.1.1 匿名转发

//发送消息
String exchange = "";
channel.basicPublish(exchange,RabbitConstant.QUEUE_NAME_WORK,null,msg.getBytes());

匿名转发：”” （exchange为空串表示匿名转发）也就是不指定转化器，也就是在普通的队列模式中使用

2.6.1.2 不处理路由（Fanout）

//声明交换机
channel.exchangeDeclare(RabbitConstant.EXCHANGE_NAME_FANOUT,"fanout");//分发

Fanout (不处理路由键)

只要绑定到交换机的队列都能获取到消息

//向交换机中推送消息
String routingKey= "";
channel.basicPublish(RabbitConstant.EXCHANGE_NAME_FANOUT,routingKey, null, msg.getBytes());

不处理路由：”” （routingKey为空串表示不处理路由）

2.6.1.3 处理路由（Direct）

Direct (处理路由键)

通过模型我们可以看到，在发送方推送消息时指定路由键的名称，在接收方的队列必须指定对应的路由键才能获得相应的消息，这就可以让我们选择性的订阅指定消息

发送方：

//声明交换机
channel.exchangeDeclare(RabbitConstant.EXCHANGE_NAME_FANOUT,"direct");//分发
//向交换机中推送消息
String msg = "hello publish subscribe !";
String routingKey= "text_routing_key";
channel.basicPublish(RabbitConstant.EXCHANGE_NAME_FANOUT,routingKey, null, msg.getBytes());

接收方：

//绑定队列到交换机
String routingKey= "text_routing_key";
channel.queueBind(RabbitConstant.QUEUE_NAME_EXCHANGE_BIND_EMIL,RabbitConstant.EXCHANGE_NAME_FANOUT,routingKey);

2.6.1.4 主题（Topic）

将路由键和某模式进行匹配

#匹配一个或多个单词

*匹配一个单词

比如：goods.*只能匹配到 goods.add , goods.# 则可以匹配到goods.add.chongqing

指定为主题模式时，可以通过#、*通配符来匹配不同的路由键

生产者：

/声明交换机
channel.exchangeDeclare(RabbitConstant.EXCHANGE_NAME_TOPIC,"topic");//分发
//向交换机中推送消息
String msg = "hello publish subscribe !";
String routingKey = "goods.add";//指定路由键
channel.basicPublish(RabbitConstant.EXCHANGE_NAME_TOPIC,routingKey, null, msg.getBytes());

接收方：

//绑定队列到交换机
String routingKey = "goods.#";//指定路由键
channel.queueBind(RabbitConstant.QUEUE_NAME_TOPIC_TWO,RabbitConstant.EXCHANGE_NAME_TOPIC,routingKey);

2.6.2 basicPublish 方法（发送消息）

void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException;

exchange：交换机的名称

routingKey：路由键，#匹配0个或多个单词，*匹配一个单词，在topic exchange做消息转发用

mandatory：true：如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者。false：出现上述情形broker会直接将消息扔掉

immediate：true：如果exchange在将消息route到queue(s)时发现对应的queue上没有消费者，那么这条消息不会放入队列中。当与消息routeKey关联的所有queue(一个或多个)都没有消费者时，该消息会通过basic.return方法返还给生产者。
BasicProperties ：需要注意的是BasicProperties.deliveryMode，0:不持久化 1：持久化这里指的是消息的持久化，配合channel(durable=true),queue(durable)可以实现，即使服务器宕机，消息仍然保留

body：推送的消息内容

简单来说：mandatory标志告诉服务器至少将该消息route到一个队列中，否则将消息返还给生产者；immediate标志告诉服务器如果该消息关联的queue上有消费者，则马上将消息投递给它，如果所有queue都没有消费者，直接把消息返还给生产者，不用将消息入队列等待消费者了。

2.7路由模式

2.7.1 模型

2.7.2 生产者

1、定义路由模式下交换机名称

public class RabbitConstant {
/**
* 简单队列使用
*/
public final static String QUEUE_NAME_SIMPLE = "test_simple_queue";
/**
* 工作队列使用
*/
public final static String QUEUE_NAME_WORK = "test_work_queue";
/**
* 发布订阅使用（不路由)
*/
public final static String EXCHANGE_NAME_FANOUT = "test_fanout_exchange";//交换机名称
public final static String QUEUE_NAME_EXCHANGE_BIND_EMIL = "test_queue_fanout_emil";//绑定到交换机的队列
public final static String QUEUE_NAME_EXCHANGE_BIND_SMS = "test_queue_fanout_sms";//绑定到交换机的队列
/**
* 发布订阅使用（路由）
*/
public final static String EXCHANGE_NAME_DIRECT = "test_direct_exchange";//交换机名称
public final static String QUEUE_NAME_DIRECT_ONE = "test_queue_direct_one";//绑定到交换机队列
public final static String QUEUE_NAME_DIRECT_TWO = "test_queue_direct_two";//绑定到交换机对列
}

2. 创建生产者

public class send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(RabbitConstant.EXCHANGE_NAME_DIRECT,"direct");//分发
//向交换机中推送消息
String msg = "hello publish subscribe !";
String routingKey = "error";//指定路由键
// routingKey = "warning";
// routingKey = "info";
channel.basicPublish(RabbitConstant.EXCHANGE_NAME_DIRECT,routingKey, null, msg.getBytes());
System.out.println("send publish subscribe success：" + msg);
//关闭通道和连接
channel.close();
connection.close();
}
}

2.7.3 消费者

队列名称的定义在上面生产者那里已经给出

消费者1：

public class Recv1 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_DIRECT_ONE,false,false,false,null);
//绑定队列到交换机
String routingKey = "error";//指定路由键
channel.queueBind(RabbitConstant.QUEUE_NAME_DIRECT_ONE,RabbitConstant.EXCHANGE_NAME_DIRECT,routingKey);
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【1】Recv success："+msg);
try {
Thread.sleep(600);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【1】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_DIRECT_ONE,autoAck,consumer);
}
}

消费者2：

public class Recv2 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_DIRECT_TWO,false,false,false,null);
//绑定队列到交换机
String routingKey = "error";//指定路由键
channel.queueBind(RabbitConstant.QUEUE_NAME_DIRECT_TWO,RabbitConstant.EXCHANGE_NAME_DIRECT,routingKey);
routingKey = "info";
channel.queueBind(RabbitConstant.QUEUE_NAME_DIRECT_TWO,RabbitConstant.EXCHANGE_NAME_DIRECT,routingKey);
routingKey = "warning";
channel.queueBind(RabbitConstant.QUEUE_NAME_DIRECT_TWO,RabbitConstant.EXCHANGE_NAME_DIRECT,routingKey);
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【2】Recv success："+msg);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【2】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_DIRECT_TWO,autoAck,consumer);
}
}

2.7.4 总结

通过消费者代码我们可以看到，消费者1只能接收到路由键是“error”的消息，消费者2可以接收到路由键是“error”、“info”、“warning”的消息。

我们通过改变生产者中的路由键名称就能指定到哪些消费者可以接收消息了。

2.8 主题模式

2.8.1 模型

Topic Exchange-将路由键和某模式进行匹配。此时队列需要绑定要一个模式上，符号"#"匹配一个或多个词，符号匹配不多不少一个词。因此" audit.# "能够匹配到"audit.irs.corporate",但是" audit.* "只会匹配到"auditirs.irs"。

2.8.2 生产者

1、定义主题模式下转换器名称

public class RabbitConstant {
/**
* 简单队列使用
*/
public final static String QUEUE_NAME_SIMPLE = "test_simple_queue";
/**
* 工作队列使用
*/
public final static String QUEUE_NAME_WORK = "test_work_queue";
/**
* 发布订阅使用（不处理路由模式【订阅模式】 Fanout)
*/
public final static String EXCHANGE_NAME_FANOUT = "test_fanout_exchange";//交换机名称
public final static String QUEUE_NAME_EXCHANGE_BIND_EMIL = "test_queue_fanout_emil";//绑定到交换机的队列
public final static String QUEUE_NAME_EXCHANGE_BIND_SMS = "test_queue_fanout_sms";//绑定到交换机的队列
/**
* 发布订阅使用（处理路由模式 Direct）
*/
public final static String EXCHANGE_NAME_DIRECT = "test_direct_exchange";//交换机名称
public final static String QUEUE_NAME_DIRECT_ONE = "test_queue_direct_one";//绑定到交换机队列
public final static String QUEUE_NAME_DIRECT_TWO = "test_queue_direct_two";//绑定到交换机对垒
/**
* 发布订阅使用（主题模式 Topic）
*/
public final static String EXCHANGE_NAME_TOPIC = "test_topic_exchange";//交换机名称
public final static String QUEUE_NAME_TOPIC_ONE = "test_queue_topic_one";//绑定到交换机队列
public final static String QUEUE_NAME_TOPIC_TWO = "test_queue_topic_two";//绑定到交换机对垒
}

2、创建生产者

public class send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(RabbitConstant.EXCHANGE_NAME_TOPIC,"topic");//分发
//向交换机中推送消息
String msg = "hello publish subscribe !";
String routingKey = "goods.add";//指定路由键
// routingKey = "goods.update";
channel.basicPublish(RabbitConstant.EXCHANGE_NAME_TOPIC,routingKey, null, msg.getBytes());
System.out.println("send publish subscribe success：" + msg);
//关闭通道和连接
channel.close();
connection.close();
}
}

2.8.3 消费者

队列名称的定义在上面生产者那里已经给出

消费者1：

public class Recv1 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_TOPIC_ONE,false,false,false,null);
//绑定队列到交换机
String routingKey = "goods.add";//指定路由键
channel.queueBind(RabbitConstant.QUEUE_NAME_TOPIC_ONE,RabbitConstant.EXCHANGE_NAME_TOPIC,routingKey);
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【1】Recv success："+msg);
try {
Thread.sleep(600);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【1】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_TOPIC_ONE,autoAck,consumer);
}
}

消费者2：

public class Recv2 {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
final Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_TOPIC_TWO,false,false,false,null);
//绑定队列到交换机
String routingKey = "goods.#";//指定路由键
channel.queueBind(RabbitConstant.QUEUE_NAME_TOPIC_TWO,RabbitConstant.EXCHANGE_NAME_TOPIC,routingKey);
//保证一次只分发一个
channel.basicQos(1);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("【2】Recv success："+msg);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("【2】 done ！");
//手动回执消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
Boolean autoAck = false;//自动应答改为false
channel.basicConsume(RabbitConstant.QUEUE_NAME_TOPIC_TWO,autoAck,consumer);
}
}

2.8.4 总结

通过消费者代码我们可以看到，消费者1只能接收到路由键是“goods.add”的消息，消费者2路由键为“goods.#”，可以接收到路由键是“goods.add”、“goods.update”、“goods.delete.tt”等含“goods.”前缀的1-n个单词组成的键的消息。

如果路由键是“goods.*”的话，只能匹配到前缀为“goods.”的一个单词的路由键，比如可以匹配“goods.add”，但是不能匹配“goods.delete.tt”。

2.9 RabiitMq的消息确认机制（事务+confirm）

在rabbitmq中我们可以通过持久化数据解决rabbitmq服务器异常的数据丢失问题

问题：生产者将消息发送出去之后,消息到底有没有到达rabbitmq服务器，默认的情况是不知道的

两种方式：
AMOP实现了事务机制
Confirm模式

2.9.1 事务机制

txSelect txCommit txRollback
txSelect：用户将当前channel设置成transation模式,
txCommit：用于提交事务
txRollback：回滚事务

2.9.1.1 生产者

public class send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE,false,false,false,null);
try {
//开启事务
channel.txSelect();
//向队列中推送消息
String msg = "hello tx message!";
channel.basicPublish("",RabbitConstant.QUEUE_NAME_SIMPLE,null,msg.getBytes());
//事务提交
channel.txCommit();
System.out.println("send tx success："+msg);
}catch (Exception e){
//事务回滚
channel.txRollback();
System.out.println("send tx message rollback !");
}
//关闭通道和连接
channel.close();
connection.close();
}
}

2.9.1.2 消费者

public class Recv {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE,false,false,false,null);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body,"utf-8");
System.out.println("Recv【tx】 success："+msg);
}
};
//监听队列
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE,true,consumer);
}
}

2.9.1.3 总结

在生产者处开启事务处理，如果发生异常则消息将不会进行提交，而是会进行回滚，那么消费者端是接收不到消息的。这个事务处理机制因为连接增加了，降低了消息服务器mq的吞吐量。

2.9.2 confirm模式

2.9.2.1 实现原理

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号,此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

Confirm最大的好处是异步处理的，不用等待信道返回确认消息，而继续发送其他消息

如果mq崩溃异常导致数据丢失会返回一条Nack消息，生产者可以利用回调进行处理

开启confirm：

channel.confirmSelect();

编程模式：

普通模式：

发一条：waitForConfirms()

批量发：waitForConfirms()

异步confirm模式：

提供一个回调方法

2.9.2.2 生产者（简单模式）

定义队列名称：

public final static String QUEUE_NAME_SIMPLE_CONFIRM = "test_simple_confirm_queue";

1、Confirm 单条：

public class send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, false, false, false, null);
//开启confirm模式
channel.confirmSelect();
//向队列中推送消息
String msg = "hello confirm message!";
channel.basicPublish("", RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, null, msg.getBytes());
//确认
if(!channel.waitForConfirms()){
System.out.println("send confirm failed：" + msg);
}else {
System.out.println("send confirm success：" + msg);
}
//关闭通道和连接
channel.close();
connection.close();
}
}

2、批量

public class SendBatch {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, false, false, false, null);
//开启confirm模式
channel.confirmSelect();
//批量向队列中推送消息
String sendMsg = "";
for(int i = 0;i<20;i++){
sendMsg = "hello confirm message{"+i+"] !";
channel.basicPublish("", RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, null, sendMsg.getBytes());
}
//确认
if(!channel.waitForConfirms()){
System.out.println("Send confirm failed：");
}else {
System.out.println("Send confirm success：");
}
//关闭通道和连接
channel.close();
connection.close();
}
}

发完消息在进行确认

2.9.2.3 生产者（异步模式）

Channel对象提供的Confirmlistener()回调方法只包含deliveryTag (当前Chanel发出的消息序号)，我们需要自己为每一个Channel维护一个unconfirm的消息序号集合，每 publish一条数据，集合中元素加1，每回调一次handleAck方法, unconfirm集合删掉相应的一条(multiple=false)或多条(multiple=true)记录。从程序运行效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构。

生产者：

public class Send {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, false, false, false, null);
//开启confirm模式
channel.confirmSelect();
//未确认的消息标识
final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
channel.addConfirmListener(new ConfirmListener(){
//没有问题的handleAck
public void handleAck(long deliveryTag, boolean multiple) {
if(multiple){
System.out.println("----handleAck----multiple");
confirmSet.headSet(deliveryTag+1).clear();
}else {
System.out.println("----handleAck----multiple false");
confirmSet.remove(deliveryTag);
}
}
//handleNack
public void handleNack(long deliveryTag, boolean multiple) {
if(multiple){
System.out.println("----handleNack----multiple");
confirmSet.headSet(deliveryTag+1).clear();
}else {
System.out.println("----handleNack----multiple false");
confirmSet.remove(deliveryTag);
}
}
});
//向队列中推送消息
for(int i = 0 ;i<50;i++){
long nextPublishSeqNo = channel.getNextPublishSeqNo();
String msg = "hello confirm message["+i+"] !";
channel.basicPublish("", RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, null, msg.getBytes());
confirmSet.add(nextPublishSeqNo);
}
//关闭通道和连接
channel.close();
connection.close();
}
}

2.9.2.4 消费者

public class Recv {
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = RabbitConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, false, false, false, null);
//重写handleDelivery 方法，实现阻塞之后获取消息的处理
Consumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//打印接收到的内容
String msg = new String(body, "utf-8");
System.out.println("Recv【confirm】 success：" + msg);
}
};
//监听队列
channel.basicConsume(RabbitConstant.QUEUE_NAME_SIMPLE_CONFIRM, true, consumer);
}
}

3、Spring集成Rabbitmq

3.1 引入maven依赖

<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>2.1.8.RELEASE</version>
</dependency>

3.2 配置application.xml文件

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit
https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd">
<rabbit:connection-factory id="connectionFactory" host="127.0.0.1" port="5672" username="guest" password="guest" virtual-host="/test_rabbit"/>
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="fanoutExchange"/>
<rabbit:admin connection-factory="connectionFactory"/>
<rabbit:queue name="myqueue" auto-declare="true" durable="true"/>
<rabbit:fanout-exchange name="fanoutExchange" auto-declare="true">
<rabbit:bindings>
<rabbit:binding queue="myqueue"/>
</rabbit:bindings>
</rabbit:fanout-exchange>
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="foo" method="listen" queue-names="myqueue"/>
</rabbit:listener-container>
<bean id="foo" class="com.rabbit.test.spring.MyConsumer"/>
</beans>

3.3 创建消费者监听

public class MyConsumer {
//具体执行的业务方法
public void listen(String foo){
System.out.println("消费者："+foo);
}
}

3.4 创建生产者发送消息

public class SpringMain {
public static void main(String[] args) throws Exception {
//获得配置上下文
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:application.xml");
//得到rabbitmq模板
RabbitTemplate rabbitTemplate = ctx.getBean(RabbitTemplate.class);
//发送消息
rabbitTemplate.convertAndSend("hello word !");
Thread.sleep(1000);
ctx.destroy();
}
}

RabbitMq 操作总结

RabbitMq java操作总结

1、添加用户

用户界面

Virtual hosts 管理

2、java操作队列

2.1简单队列

2.1.1 模型

2.2 工作队列（轮询分发 round-robin）

2.3 工作队列（公平分发fail dipatch）

2.3.1 生产者

2.4 消息应答与消息持久化

2.5 订阅模式（publish_subscribe）

2.6 exchange (交换机 转换器)

2.7路由模式

2.8 主题模式

2.9 RabiitMq的消息确认机制（事务+confirm）

3、Spring集成Rabbitmq

3.1 引入maven依赖

3.2 配置application.xml文件

3.3 创建消费者监听

3.4 创建生产者发送消息

2.6 exchange (交换机转换器)