RabbitMq开发向导之生产者(进阶)
文章目录
mandatory参数
//交换器与队列通过路由键绑定
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
/**
* 发送消息到队列中
* mandatory为true时,交换器无法根据自身的类型和路由键找到符合条件的队列,那么RabbitMQ会调用Basic.return将消息返回生产者。
* false时,消息直接被丢弃
*/
channel.basicPublish(EXCHANGE_NAME, "", true, MessageProperties.PERSISTENT_TEXT_PLAIN, "mandatory test".getBytes("UTF-8"));
channel.addReturnListener(new ReturnListener() {
@Override
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg =new String(body);
System.out.println("Basic.Return----->"+msg);
}
});
当mandatory参数为true时,交换器如果无法根据类型和路由键找到符合条件的队列,那么RabbitMq会调用Basic.Return命令将消息返回给生产者
备份交换器
/**
* 创建一个备份参数
*/
Map<String,Object> params=new HashMap<String,Object>();
params.put("alternate-exchange","myAe");
//设置备份转换器
channel.exchangeDeclare("normalEx","direct",true,false,params);
channel.exchangeDeclare("myAe2","direct",true,false,null);
channel.queueDeclare("normalQ",true,false,false,null);
channel.queueBind("normalQ","normalEx","nomalKey");
channel.queueDeclare("unroutedQ",true,false,false,null);
channel.queueBind("unroutedQ","myAe2","aeKey");
channel.basicPublish("normalEx","aeKey",MessageProperties.PERSISTENT_TEXT_PLAIN,"hello".getBytes("UTF-8"));
运行结果:
代码中申明了两个交换器normalEx和myAe2,并且类型都是direct,通过参数的声明,normalEx的交换器备份交换器为myAe2,两个交换器各自绑定自己的队列normalQ和unroutedQ,routingKey分别为nomalKey和aeKey,当消息发送给交换器normalEx但是routingKey和备份交换器匹配时,消息则进入备份交换器的队列
几种特殊情况说明
- 1.如果设置的备份交换器不存在,客户端和RabbitMq服务端都不会有异常出现,此时消息会丢失
- 2.如果备份交换器没有绑定任何队列,客户端和RabbitMq服务端都不会有异常出现,此时消息会丢失
- 3.如果备份交换器没有任何匹配的队列,客户端和RabbitMq服务端都不会有异常出现,此时消息会丢失
- 4.备份交换器和mandatory参数一起用时,mandatory参数无效
过期时间(TTL)
队列过期时间
Map<String,Object> params=new HashMap<String,Object>();
params.put("x-message-ttl",6000);
params.put("x-expires",6000);
//申明队列---->整个队列TTL过期时间
channel.queueDeclare(GlobalConstants.QUEUE_NAME,true,false,false,params);
队列的过期时间参数有两种
x-message-ttl:过期时,消息在队列中“死掉”
x-expires:控制 queue 被自动删除前可以处于“未使用”状态的时间。“未使用”的意思是指 queue 上没有任何消费者 consumer ,且在过期时间段内 queue 没有被重新声明,也未通过 basic.get 命令被访问过。
消息过期时间
channel.basicPublish(null,null,new AMQP.BasicProperties.Builder().deliveryMode(2).expiration("60000").build(),"ttlMsg".getBytes("UTF-8"));
消息过期时间直接在发送消息时使用函数设值,注意,消息不是到了过期时间就在队列中消失,而是在即将投递到消费者之前判定的,当消费者消费此消息时,判断是否过期
死信队列(DLX)/ 延迟队列
channel.exchangeDeclare("exchange.dlx","direct",false);
channel.exchangeDeclare("exchange.normal","fanout",false);
Map<String,Object> argMap=new HashMap<String,Object>();
argMap.put("x-message-ttl",10000);
argMap.put("x-dead-letter-exchage","exchange.dlx");
argMap.put("x-dead-letter-routing-key","dlxKey");
channel.queueDeclare("queue.normal",false,false,false,argMap);
channel.queueBind("queue.normal","exchange.normal","");
channel.queueDeclare("queue.dlx",false,false,false,null);
channel.queueBind("queue.dlx","exchange.dlx","dlxKey");
channel.basicPublish("exchange.normal","dlxKey", MessageProperties.PERSISTENT_TEXT_PLAIN,"dlx".getBytes("UTF-8"));
首先,将消息发送给normal队列,并且设置过期时间,但到过期时间还没有消费者消费时,则通过设置的x-dead-letter-exchange和x-dead-letter-routing-key参数,将消息发送到对应的死信队列
延迟队列
当用户想指定时间去操作一些事情是,可以使用延迟队列
RabbitMq本身没有延迟队列的api,可以通过死信队列的方式模拟出延迟队列
生产者确认
事务机制
try{
channel.txSelect();
channel.basicPublish(GlobalConstants.EXCHANGE_NAME, GlobalConstants.ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes("UTF-8"));
channel.txCommit();
}catch (Exception e){
channel.txRollback();
}
消息成功被RabbitMq接收,事务才能提交成功,否则可以捕获异常后回滚,同时可以进行消息的重发,但是事务机制很吃RabbitMq性能
发送方确认机制
//消息确认机制
try{
channel.confirmSelect();
channel.basicPublish(GlobalConstants.EXCHANGE_NAME, GlobalConstants.ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes("UTF-8"));
if(!channel.waitForConfirms()){
System.out.println("msg send faild");
//dosomething
}
}catch (Exception e){
}
当信道进入confirm模式时,该信道上发布消息会指派一个唯一的id,一旦消息被投递到所有匹配队列后,RabbitMq就会发送一个确认Basic.Ack给生产者(包含消息的唯一id),这就使得生产者知晓消息已经正确到达目的地。如果消息和队列是持久化的,那么确认消息会在消息写入磁盘后发出。
优势
- 可以批量confirm,发送一批消息后再调用channel.waitForConfirms()方法
- 异步confirm,提供一个回调方法,在消息确认后客户端回调这个方法进行处理
注意要点
- 事务机制和确认机制是互斥的,不能同时存在
- 两个机制确保的是消息能正确的发往RabbitMq交换器,如果交换器没有匹配的队列,消息也会消失