消息中间件Rabbit Mq的了解与使用
MQ（消息队列）作为现代比较流行的技术，在互联网应用平台中作为中间件，主要解决了应用解耦、异步通信、流量削锋、服务总线等问题，为实现高并发、高可用、高伸缩的企业应用提供了条件。
目前市面比较流行的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等，而每种产品又有着独特的着重点，可根据业务需要进行选择。
这里有对主流MQ的优缺点的一些描述与比较。
- 安装
1、下载Erlang
官网下载，安装完成配置环境变量
2、安装RabbitMq
官网下载，直接安装
启用管理界面：rabbitmq-plugins.bat enable rabbitmq_management
进入%RM%/sbin双击 rabbitmq-server.bat，启动成功后如下：
启动完成浏览器访问http://localhost:15672，用户名/密码：guest/guest，进入界面后如下：
- 介绍
RabbitMq是遵循了AMQP协议的开源消息队列的代理服务软件。
- 跨平台，支持多种语言
- 实现了AMQP协议；
- 满足高并发需求
- 支持集群部署
- 支持多插件，可视化视图
- 社区活跃
- 等等
核心元件包括：
ConnectionFactory（连接管理器）：应用程序与Rabbit之间建立连接的管理器，程序代码中使用；
Channel（信道）：消息推送使用的通道；
Exchange（交换器）：用于接受、分配消息；
Queue（队列）：用于存储生产者的消息；
RoutingKey（路由键）：用于把生成者的数据分配到交换器上；
BindingKey（绑定键）：用于把交换器的消息绑定到队列上；
- 使用
先看一段简单的代码，来理解下消息队列的工作原理：
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
final Channel channel = connection.createChannel();
channel.queueDeclare("q.demo",false,false,true,null);
channel.basicPublish("","q.demo",null,"消息".getBytes());
channel.basicConsume("q.demo",false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body));
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
- 首先获取连接，之前下载的rabbitmq server，与之建立连接
- 打开通信channel，可以理解为打开一个tcp通信，只不过为节省资源虚拟了一个通道
- 定义一个队列
- 向队列中发送消息
- 从队列中消费消息
其实这里还有个过程被忽略了，其实是使用了默认处理，在第4步中，其实是向消息路由发布消息，且该消息路由的routingKey与队列名称相同，
The default exchange is implicitly bound to every queue, with a routing key equal to the queue name. It is not possible to explicitly bind to, or unbind from the default exchange. It also cannot be deleted.
而实际中的用法可能更为复杂些，但是 原理上其实都一样，只不过在消息的处理过程中会添加一些策略来应对不同的应用场景,同时为了保证消息的可靠性，会引入一些确认机制，当然这些都是后话，先在刚的基础上看一下另一个示例，同样有生产者-消费作为模型：
生成者：
public void produce(String message) throws IOException, TimeoutException, InterruptedException {
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
//基于信道的通信
Channel channel = connection.createChannel();
/**
* 交换机名称、交换机类型、是否持久化、是否自动删除、是否内部使用、参数
*/
channel.exchangeDeclare(Common.EXCHANGE_X1,ExType.Direct.value(),false,false,false,null);//申明交换机
/**
* 消息发送到指定交换机、routing key、是否重发、是否、基础属性、消息内容
* mandatory:(true)没有队列，消息返回;(false)没有队列，消息丢弃
* immediate:(true)没有消费者，消息返回;(false)
*/
int count = 0;
while (count++ <100){
TimeUnit.SECONDS.sleep(1);
channel.basicPublish(Common.EXCHANGE_X1,Common.ROUTING_KEY1,false,false,null,(message+ new Date()).getBytes());
}
channel.close();
AmqpConnectionFactory.close(connection);
}
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
Producer producer = new Producer();
producer.produce("消息 ");
}
大致意思就是讲生产的消息绑定到指定的交换机上，并且指定交换机类型以及routing key。
消费者：
public void consume() throws IOException {
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
final Channel channel = connection.createChannel();//建立通信通道
/**
* 队列名称、是否持久化、是否被该连接独占、自动删除、参数
*/
channel.queueDeclare(Common.QUEUE_Q1,false,false,false,null);//申明队列
/**
* 交换机名称、交换机类型、是否持久化、是否自动删除、是否内部使用、参数
*/
channel.exchangeDeclare(Common.EXCHANGE_X1,ExType.Direct.value(),false,false,false,null);//申明交换机
/**
* 队列名称、交换机名称、binding Key
*/
channel.queueBind(Common.QUEUE_Q1,Common.EXCHANGE_X1,Common.BINDING_KEY1);//将消息交换机与队列绑定
/**
* 队列名称、自动ACK、消费者标记、非本地、是否被该连接独占、参数
*/
channel.basicConsume(Common.QUEUE_Q1,false,"c1",false,false,null,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println(new String(body));
/**
* 消息索引、批量确认
*/
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
public static void main(String[] args) throws IOException {
Consumer consumer = new Consumer();
consumer.consume();
}
相对要复杂些，主要是从队列中读取消息并显示，那么队列的消息则是通过交换机与设置的binding key来决定。
现在来梳理下整个流程：
- 在生产者中建立与mq服务的连接，创建通道
- 定义消息交换机，注意次数有很多参数，现在我们仅关注其名称与类型
- 循环100次向指定交换机中发布消息，并设置routing key
- 在消费者中建立连接，创建通道
- 定义消息队列
- 定义消息交换机
- 将消息队列绑定到交换机上，并定义binding key
- 从队列中读取消息
其中有些方面需要注意的：
- 生产者与消费者启动不分先后
- 两个地方都定义了消息交换机，与上一个对应。因为如果有该交换已经存在，则不会重新创建，但如果属性不同，则会报错
- 其中涉及到了routing key、binding key，主要是完成消息从匹配的交换机拿取消息，当然这个不是必须的，因为受到exchange的type影像
- 生产者的的消息都是发送到交换机，而消费的消息都是从队列中拿
到这里有必要说下Exchange的type，主要有以下类型：
- Fanout：转发消息到所有绑定队列
- Direct：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing key, 消息的routing key 匹配时, 才会被交换器投送到绑定的队列中去.
- Topic：按规则转发消息（最灵活）
- Headers：设置 header attribute 参数类型的交换机
通过这几种类型的exchange，可以定制出非常复杂的业务模型。
上面可以说从简单的应用层面了解了Rabbit Mq，因为网上有太多的知识，对一些组件与工作模型等都讲解，并且图文并茂，所有没必要做过多重复的工作，下面从可靠性的角度来学习，同时之前我们在声明队列、交换机等会有一些参数，具体干嘛用的，可能有些会说到，毕竟是学习阶段，不可能那么全面，也不会那么准确，权作参考。
下面看一个简单的im，仅仅实现两个人间的固化通信，jack and rose：
首先说下思路，两个用户间的通信，会涉及到每个用户的接收和发送，我们知道，一般聊天时，接收和发送是互不影响的，也不会有任何依赖关系，也就是发送和接收的消息是两条线，那么我们就需要为每个用户分别开通一个发送和接收的线程，这样两个行为就不会有任何影响。然后看下怎么发送信息，就是通过mq开通一个channel，将消息发送到对应的exchange，进而讲消息推送到匹配的消息队列中，而另一方接收，则从指定的队列中取得消息并展现出来。那么接收发送就形成了完整的过程。当然这个和之前说的接收发送不同，刚才指的是同一用户，现在指用户到用户，这个发送与接收本就是一个过程的两个阶段。下面看下代码实现，既然两个用户的行为如此相像，我们就提出一个抽象类来实现共同的部分：
public abstract class AbstractUser {
private static String EXCHANGE = "x.user";
private String id = UUID.randomUUID().toString();
private String name;
private Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
public AbstractUser(String name){
this.name = name;
}
public void start() throws IOException {
System.out.println(name +" 上线了..");
Channel channel = connection.createChannel();
channel.basicQos(1);//流量控制
String queueName = getResQueue();
channel.exchangeDeclare(EXCHANGE,ExType.Direct.value());
channel.queueDeclare(queueName,false,false,false,null);
channel.queueBind(queueName,EXCHANGE,getResBindingKey());
//专门接收的线程
new Thread(()->{
try {
channel.basicConsume(queueName,false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if(!id.equals(properties.getCorrelationId())){
System.out.println(properties.getAppId()+" : "+new String(body));
channel.basicAck(envelope.getDeliveryTag(),false);
}else{
channel.basicNack(envelope.getDeliveryTag(),false,true);
}
}
});
} catch (IOException e) {
e.printStackTrace();
}
}).start();
//专门发送的线程
new Thread(()->{
Scanner scanner = new Scanner(System.in);
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().correlationId(id).appId(name).build();
while (true){
try {
channel.basicPublish(EXCHANGE,getPusRoutingKey(),properties,scanner.nextLine().getBytes());
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
}
/**
* 发送消息时指定routing key，在接收方binding key需要与之对应
* @return
*/
public abstract String getPusRoutingKey();
/**
* 接收消息的队列
* @return
*/
public abstract String getResQueue();
/**
* 接收消息的binding key，与发送方的routing key对应
* @return
*/
public abstract String getResBindingKey();
}
JACK:
public class Jack extends AbstractUser{
public Jack(String name) {
super(name);
}
@Override
public String getPusRoutingKey() {
return "toRose";
}
@Override
public String getResQueue() {
return "q.rose";
}
@Override
public String getResBindingKey() {
return "toJack";
}
public static void main(String[] args) throws IOException {
Jack jack = new Jack("jack");
jack.start();
}
}
ROSE:
public class Rose extends AbstractUser{
public Rose(String name) {
super(name);
}
@Override
public String getPusRoutingKey() {
return "toJack";
}
@Override
public String getResQueue() {
return "q.jack";
}
@Override
public String getResBindingKey() {
return "toRose";
}
public static void main(String[] args) throws IOException {
Rose rose = new Rose("rose");
rose.start();
}
}
其中要注意的是，在接收的时候，开始设计时是共用了一个队列，所以会出现自己给自己发信息，所以在发送消息时，为消息添加了属性，标识该消息的来源，那么在读取消息时，根据该属性判断是否为自己的消息，如果是，则确认并消费该消息，如果不是，需要做一次nack的处理，并将消息重新放回队列中，直到被其他用户消费为止。我们可以看到，现在是两个人的通信，有一些固化的元素，比如routing key，两个用户通信是需要优先确定的，那么真实的IM系统，会涉及到很多繁琐的内容，比如消息发送失败，消息发送超时、重发、多人聊天等等，会存在很多需要解决的问题。
jack和rose的聊天也结束了，那么我们在来看看其他的一些知识点，同样以消息的发送与消息接收为一条线来进行下去。
在发送消息前，毫无疑问是先建立连接，打开虚拟通道，之后才是定义交换机，发送消息（不用申明队列）。那么在申明交换机的时候，其实有很多个参数：
Exchange.DeclareOk exchangeDeclare(String exchange,
String type,
boolean durable,
boolean autoDelete,
boolean internal,
Map<String, Object> arguments) throws IOException;
- exchange：交换机名称
- type：交换机类型，上面说到了， direct, topic, fanout, headers
- durable：是否持久化，也就是断开连接后是否还存在
- autoDelete：自动删除，当与该exchange上的队列全部删除后, 自动删除，和上一个参数比较一下，比如durable=true，那么如果该参数配置true，其实也会删除（没有queue）
- internal：是否内部交换机，不太知道应用场景
- arguments：其他参数，比如DLX
在发送消息时，同样有一些可配置参数：
void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
throws IOException;
- exchange：消息发送到的交换机
- routingKey：交换机路由key
- mandatory：true，如果交换机没有匹配到对应的队列，会将调用basic.return将该消费返回生成者；false，上述情形直接丢弃消息
- immediate：true，如果交换机关联队列没有消费者，则不会将消息加入队列；false，上述情形将调用basic.return将消息返回生产者。3.0后去掉了
- props：为消息添加一些参数，比如过期时间
- body：消息主体
那么这些参数主要干嘛的？当时是保证系统的可靠性了。
那么在消息的发送端，如何保证可靠性：
- 事务
try {
channel.txSelect();
channel.basicPublish("exchange", "routingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, msg);
channel.txCommit();
} catch (Exception ex) {
channel.txRollback();
}
- 确认机制（推荐）
该机制主要是通过注册一些事件来处理的，比如上面提到过的basic.return
channel.confirmSelect();
channel.basicPublish("yu.exchange", "yu.1", MessageProperties.PERSISTENT_TEXT_PLAIN, msg);
boolean success = channel.waitForConfirms(10);
channel.addConfirmListener(new ConfirmListener() {
public void handleAck(long l, boolean b) throws IOException {
}
public void handleNack(long l, boolean b) throws IOException {
}
});
channel.addReturnListener(new ReturnListener() {
public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException{
}
});
channel.confirmSelect();
channel.basicPublish("exchange", "routingKey",true, MessageProperties.PERSISTENT_TEXT_PLAIN, msg);
下面主要对第二种情况验证一下，记得在测试前，讲相关的exchange与queue进行删除，否则会影响测试结果：
public class Producer {
public void produce(String message) throws IOException, TimeoutException, InterruptedException {
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
//基于信道的通信
Channel channel = connection.createChannel();
/**
* 交换机名称、交换机类型、是否持久化、是否自动删除、是否内部使用、参数
*/
channel.exchangeDeclare(Common.EXCHANGE_X1,ExType.Direct.value(),false,false,false,null);//申明交换机
channel.confirmSelect();//确认机制
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息发送成功！");
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息发送失败！");
}
});
/**
* mandatory=ture
*/
channel.addReturnListener(new ReturnListener() {
@Override
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("重新处理消息！");
}
});
/**
* 消息发送到指定交换机、routing key、是否重发、是否、基础属性、消息内容
* mandatory:(true)没有队列，消息返回;(false)没有队列，消息丢弃
* immediate:(true)没有消费者，消息返回;(false)
*/
int count = 0;
while (count++ <10){
TimeUnit.SECONDS.sleep(1);
channel.basicPublish(Common.EXCHANGE_X1,Common.ROUTING_KEY1,false,false,null,(new Date()+message).getBytes());
}
channel.close();
AmqpConnectionFactory.close(connection);
}
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
Producer producer = new Producer();
producer.produce("消息 ");
}
}
我们发送10条消息到交换机，控制台打印如下，如果关闭连接可能最后一条消息打印不出来：
然后启动消费者，刚好也消费了10条消息（需要先声明下队列，不然生产的消息都会被丢弃，mandatory=false）。
现在我们做一些修改，将上面说到的修改mandatory=true，也就是没有与交换机匹配的队列时，将会重发，也就是调用上面我们定义的ReturnListener：
与预期的一样，只不过会发现，在调用handleReturn后会再次调用handleAck，也就是发送成功！
上面说的这些也就是消息发布者的ack机制。
接下来看下消费者的ack：
我们定义消费者时，一般会先定义队列、交换机、将队列与交换机绑定、发送消息。
声明队列：
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
Map<String, Object> arguments) throws IOException;
- queue：队列名称
- durable：是否持久化
- exclusive：排他队列，与该连接绑定，决定多个消费者是否可以访问这一队列
- autoDelete：自动删除，没有消费者时自动删除
- arguments：队列参数，比如队列过期时间
消息接收：
String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException;
- queue：绑定的队列名
- autoAck：自动ack
- consumerTag：消费者标记
- noLocal：ture，不将同一连接中生产的消息传递给该消费者
- exclusive：排他
- arguments：扩展参数
- callback：消费回调
首先要知道，消息一段被消费，就会被移除，那么我们如何确定消息是否被真实消费？因为从拿到消息到正则消掉该消息，都是有一个过程，可能任何环境都出现问题，但是被认为消费而导致消息被移除，则可靠性就无法得到保证，作为消费者和生产者一样会有事务与ack两种方式保证，只不过需要注意的是：
- autoAck=false手动应对的时候是支持事务的，也就是说即使你已经手动确认了消息已经收到了，但在确认消息会等事务的返回解决之后，在做决定是确认消息还是重新放回队列，如果你手动确认现在之后，又回滚了事务，那么已事务回滚为主，此条消息会重新放回队列；
- autoAck=true如果自定确认为true的情况是不支持事务的，也就是说你即使在收到消息之后在回滚事务也是于事无补的，队列已经把消息移除了；
那么针对ack机制，主要有以下相关方法：
//消息索引,批量ack，只对小于该DeliveryTag的消息 // 批量确认 channel.basicAck(envelope.getDeliveryTag(),false); //其中 deliveryTag 和 requeue 的含义可以参考 basicReject 方法。 multiple 参数 //设置为 false 则表示拒绝编号为 deliveryT坷的这一条消息，这时候 basicNack 和 basicReject 方法一样; // multiple 参数设置为 true 则表示拒绝 deliveryTag 编号之前所 有未被当前消费者确认的消息。 channel.basicNack(envelope.getDeliveryTag(),false,false); //一次只能拒绝一条 //其中 deliveryTag 可以看作消息的编号 ，它是一个 64 位的长整型值，最大值是 9223372036854775807 // requeue 参数设置为 true，则 RabbitMQ 会重新将这条消息存入队列，以便可以发送给下一个订阅的消费者; // requeue 参数设置为 false，则 RabbitMQ 立即会把消息从队列中移除，而不会把它发送给新的消费者 channel.basicReject(envelope.getDeliveryTag(),false);
在结束ack前，需要说明一点的是，消费者和生产者消息发送的成功与消费是否成功，并不是消费者向生产者进行ack，而是针对mq服务器。对于生产者只是确保消息发送到服务器是否成功；对于消费者，只是确保消息是否从服务器被消费掉。
如果我们对某条消息nack，有没有requeue，那么这条消息是不是真的就丢失了呢？这里不得不引入另外一个概念，死信，那么与死信对应的有死信队列XLD，同时死信的条件不只刚说到的，在以下情况都会触发：
- 消息被拒绝（basic.reject/ basic.nack）并且requeue=false
- 消息TTL过期
- 队列达到最大长度
关于第一点不多说，上面已经提到了，关于第二点TTL（time to live）关系到消息的过期时间，一般会从两个角度分析，我们知道，消息没有消费前是在队列中，那么队列的过期时间也会影响消息的过期时间，所有这个时间会从队列过期时间（）消息过期时间中取小。
队列过期时间设置：
//申明队列时设置
args.put("x-expires", 10000);//ms 队列过期时间
消息过期时间设置：
//申明队列时设置
args.put("x-message-ttl", 6000);//消息过期时间
//发布消息时设置
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("6000").build();
首先如何定义死信队列，这个是在队列申明的时候，以参数的形式加入的（x-dead-letter-exchange）：
channel.exchangeDeclare(Common.EXCHANGE_DLX_X1,ExType.Direct.value());
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange",Common.EXCHANGE_DLX_X1);//死信交换机
channel.queueDeclare(Common.QUEUE_Q1,false,false,false,args);
下面来看下示例：
消息生产者：
public void produce(String message) throws IOException, TimeoutException, InterruptedException {
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
//基于信道的通信
Channel channel = connection.createChannel();
/**
* 交换机名称、交换机类型、是否持久化、是否自动删除、是否内部使用、参数
*/
channel.exchangeDeclare(Common.EXCHANGE_X1,ExType.Direct.value(),false,false,false,null);//申明交换机
//为了保证先启动该类，交换机没有绑定队列导致消息丢失，优先处理，在消费者中也会有以下内容
channel.exchangeDeclare(Common.EXCHANGE_DLX_X1,ExType.Direct.value());
channel.queueDeclare(Common.QUEUE_DLX_Q1, false, false, false, null);//申明死信队列
channel.queueBind(Common.QUEUE_DLX_Q1, Common.EXCHANGE_DLX_X1, Common.ROUTING_DLX_KEY1);//将消息交换机与队列绑定
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange",Common.EXCHANGE_DLX_X1);//死信交换机
args.put("x-dead-letter-routing-key", Common.ROUTING_DLX_KEY1);
args.put("x-expires", 30000);//ms 队列过期时间
args.put("x-message-ttl", 12000);//消息过期时间
channel.queueDeclare(Common.QUEUE_Q1,false,false,false ,args);
channel.queueBind(Common.QUEUE_Q1,Common.EXCHANGE_X1,Common.BINDING_KEY1);//将消息交换机与队列绑定
channel.confirmSelect();//确认机制
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息发送成功！");
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息发送失败！");
}
});
/**
* mandatory=ture
*/
channel.addReturnListener(new ReturnListener() {
@Override
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("重新处理消息！");
}
});
/**
* 消息发送到指定交换机、routing key、是否重发、是否、基础属性、消息内容
* mandatory:(true)没有队列，消息返回;(false)没有队列，消息丢弃
* immediate:(true)没有消费者，消息返回;(false)
*/
int count = 0;
// AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("12000").build();
while (count++ <10){
TimeUnit.MILLISECONDS.sleep(500);
channel.basicPublish(Common.EXCHANGE_X1,Common.ROUTING_KEY1,false,false,null,(new Date()+message).getBytes());
}
channel.close();
AmqpConnectionFactory.close(connection);
}
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
Producer producer = new Producer();
producer.produce("消息 ");
}
消息消费者：
public void consume() throws IOException {
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
final Channel channel = connection.createChannel();//建立通信通道
channel.exchangeDeclare(Common.EXCHANGE_DLX_X1,ExType.Direct.value());
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange",Common.EXCHANGE_DLX_X1);//死信队列
args.put("x-dead-letter-routing-key", Common.ROUTING_DLX_KEY1);//死信routing key 默认取
args.put("x-expires", 30000);//ms 队列过期时间
args.put("x-message-ttl", 12000);//消息过期时间
/**
* 队列名称、是否持久化、是否被该连接独占(只对申明连接可见，断开连接删除)、自动删除、参数
*/
channel.queueDeclare(Common.QUEUE_Q1,false,false,false,args);//申明队列
/**
* 交换机名称、交换机类型、是否持久化、是否自动删除、是否内部使用、参数
*/
channel.exchangeDeclare(Common.EXCHANGE_X1,ExType.Direct.value(),false,false,false,null);//申明交换机
/**
* 队列名称、交换机名称、binding Key
*/
channel.queueBind(Common.QUEUE_Q1,Common.EXCHANGE_X1,Common.BINDING_KEY1);//将消息交换机与队列绑定
/**
* 队列名称、自动ACK、消费者标记、非本地、是否被该连接独占、参数
* 与basicGet对比，get 只取了队列里面的第一条消息
* 一种是主动去取，一种是监听模式
*/
channel.basicConsume(Common.QUEUE_Q1,false,"c1",false,false,null,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body));
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
public static void main(String[] args) throws IOException {
Consumer consumer = new Consumer();
consumer.consume();
}
死信消费者：
public void consume() throws IOException {
Connection connection = AmqpConnectionFactory.getConnection(new AmqpConfig());
final Channel channel = connection.createChannel();//建立通信通道
channel.queueDeclare(Common.QUEUE_DLX_Q1, false, false, false, null);//申明队列
channel.exchangeDeclare(Common.EXCHANGE_DLX_X1, ExType.Direct.value(), false, false, false, null);//申明交换机
channel.queueBind(Common.QUEUE_DLX_Q1, Common.EXCHANGE_DLX_X1, Common.ROUTING_DLX_KEY1);//将消息交换机与队列绑定
channel.basicConsume(Common.QUEUE_DLX_Q1, true, "c2", false, false, null, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("死信："+new String(body));
}
});
}
public static void main(String[] args) throws IOException {
DlxConsumer consumer = new DlxConsumer();
consumer.consume();
}
测试过程中可观察rabbitmq服务台数据变化，主要步骤大致如下：
- 执行消费者，测试会生成10条信息在正常的队列中
- 在12秒内执行消息消费者，发现会打印出10条信息
- 执行步骤1
- 过12秒后执行消息消费者，发现不会打印任何信息
- 执行死信消费者，发现打印出10条信息
至此，关于rabbitmq的知识也差不多了，但是如果想搭建一个比较稳健的消息系统来处理系统中的各种异步任务，还是需要将各种知识进行搭配。
扩展：
- 延迟队列：通过死信实现，其实上面的例子中，去掉消费者，将死信消费者看做正常消费者，那么就是延迟队列了
- 重试机制：包含发送失败重试与消费故障重试
- 队列属性：
Message TTL(x-message-ttl)：设置队列中的所有消息的生存周期(统一为整个队列的所有消息设置生命周期), 也可以在发布消息的时候单独为某个消息指定剩余生存时间,单位毫秒, 类似于redis中的ttl，生存时间到了，消息会被从队里中删除，注意是消息被删除，而不是队列被删除， 特性Features=TTL, 单独为某条消息设置过期时间AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder().expiration(“6000”); Auto Expire(x-expires): 当队列在指定的时间没有被访问(consume, basicGet, queueDeclare…)就会被删除,Features=Exp Max Length(x-max-length): 限定队列的消息的最大值长度，超过指定长度将会把最早的几条删除掉， 类似于mongodb中的固定集合，例如保存最新的100条消息, Feature=Lim Max Length Bytes(x-max-length-bytes): 限定队列最大占用的空间大小， 一般受限于内存、磁盘的大小, Features=Lim B Dead letter exchange(x-dead-letter-exchange)： 当队列消息长度大于最大长度、或者过期的等，将从队列中删除的消息推送到指定的交换机中去而不是丢弃掉,Features=DLX Dead letter routing key(x-dead-letter-routing-key)：将删除的消息推送到指定交换机的指定路由键的队列中去, Feature=DLK Maximum priority(x-max-priority)：优先级队列，声明队列时先定义最大优先级值(定义最大值一般不要太大)，在发布消息的时候指定该消息的优先级， 优先级更高（数值更大的）的消息先被消费 Lazy mode(x-queue-mode=lazy)： Lazy Queues: 先将消息保存到磁盘上，不放在内存中，当消费者开始消费的时候才加载到内存中Master locator(x-queue-master-locator)
示例代码：http://https://github.com/suspring/rabbitmq-demo
