RabbitMQ的使用
安装:Docker 安装 RabbitMQdocker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management4369,25672(Erlang发现&集群端口)5672,5671(AMQP端口)1
安装:
Docker 安装 RabbitMQ
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management
4369,25672(Erlang发现&集群端口)
5672,5671(AMQP端口)
15672 (web管理后台端口)
61613,61614(STOMP协议端口)
1883,8883(MQTT协议端口)
RabbitMQ随docker自动启动
docker update rabbitmq --restart=always
配置
设置传送rabbitmq格式化,
@Configuration
public class MyRabbitConfig {
private RabbitTemplate rabbitTemplate;
@Primary
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setMessageConverter(messageConverter());
initRabbitTemplate();
return rabbitTemplate;
}
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
/**
* 定制RabbitTemplate
* 1、服务收到消息就会回调
* 1、spring.rabbitmq.publisher-confirms: true
* 2、设置确认回调
* 2、消息正确抵达队列就会进行回调
* 1、spring.rabbitmq.publisher-returns: true
* spring.rabbitmq.template.mandatory: true
* 2、设置确认回调ReturnCallback
*
* 3、消费端确认(保证每个消息都被正确消费,此时才可以broker删除这个消息)
*
*/
// @PostConstruct //MyRabbitConfig对象创建完成以后,执行这个方法
public void initRabbitTemplate() {
/**
* 1、只要消息抵达Broker就ack=true
* correlationData:当前消息的唯一关联数据(这个是消息的唯一id)
* ack:消息是否成功收到
* cause:失败的原因
*/
//设置确认回调
rabbitTemplate.setConfirmCallback((correlationData,ack,cause) -> {
System.out.println("confirm...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]");
});
/**
* 只要消息没有投递给指定的队列,就触发这个失败回调
* message:投递失败的消息详细信息
* replyCode:回复的状态码
* replyText:回复的文本内容
* exchange:当时这个消息发给哪个交换机
* routingKey:当时这个消息用哪个路邮键
*/
rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> {
System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]" +
"==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]");
});
}
}
解决消息丢失
1、消息丢失
消息发送出去,由于网络问题没有抵达服务器
做好容错方法(try-catch),发送消息可能会网络失败,失败后要有重试机 制,可记录到数据库,采用定期扫描重发的方式
做好日志记录,每个消息状态是否都被服务器收到都应该记录 • 做好定期重发,如果消息没有发送成功,定期去数据库扫描未成功的消息进 行重发
消息抵达Broker,Broker要将消息写入磁盘(持久化)才算成功。此时Broker尚 未持久化完成,宕机。
publisher也必须加入确认回调机制,确认成功的消息,修改数据库消息状态。
自动ACK的状态下。消费者收到消息,但没来得及消息然后宕机
一定开启手动ACK,消费成功才移除,失败或者没来得及处理就noAck并重 新入 队
代码方法:
# 开启发送端消息抵达Broker确认
spring.rabbitmq.publisher-confirms=true
# 开启发送端消息抵达Queue确认
spring.rabbitmq.publisher-returns=true
# 只要消息抵达Queue,就会异步发送优先回调returnfirm
spring.rabbitmq.template.mandatory=true
# 手动ack消息,不使用默认的消费端确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
手动接受代码basicAck
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
ture为接受后删除,false为接受后不删除
拒收basicReject
channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
ture为拒收后重新入队,false为拒收后丢弃
解决消息重复:
• 2、消息重复 • 消息消费成功,事务已经提交,ack时,机器宕机。导致没有ack成功,Broker的消息 重新由unack变为ready,并发送给其他消费者
• 消息消费失败,由于重试机制,自动又将消息发送出去
• 成功消费,ack时宕机,消息由unack变为ready,Broker又重新发送
消费者的业务消费接口应该设计为幂等性的。比如扣库存有 工作单的状态标志
• 使用防重表(redis/mysql),发送消息每一个都有业务的唯 一标识,处理过就不用处理 •rabbitMQ的每一个消息都有redelivered字段,可以获取是否 是被重新投递过来的,而不是第一次投递过来的
解决消息积压:
• 3、消息积压 • 消费者宕机积压 • 消费者消费能力不足积压 • 发送者发送流量太大
• 上线更多的消费者,进行正常消费
• 上线专门的队列消费服务,将消息先批量取出来,记录数据库,离线慢慢处
简单使用
设计队列流程图
只要使用@bean注解交给spring管理即可,使用全量构造方法创建
@Configuration
public class MyRabbitMQConfig {
/* 容器中的Queue、Exchange、Binding 会自动创建(在RabbitMQ)不存在的情况下 */
/**
* 死信队列
*
* @return
*/@Bean
public Queue orderDelayQueue() {
/*
Queue(String name, 队列名字
boolean durable, 是否持久化
boolean exclusive, 是否排他
boolean autoDelete, 是否自动删除
Map<String, Object> arguments) 属性
*/
HashMap<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", "order-event-exchange");
arguments.put("x-dead-letter-routing-key", "order.release.order");
arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
Queue queue = new Queue("order.delay.queue", true, false, false, arguments);
return queue;
}
/**
* 普通队列
*
* @return
*/
@Bean
public Queue orderReleaseQueue() {
Queue queue = new Queue("order.release.order.queue", true, false, false);
return queue;
}
/**
* TopicExchange
*
* @return
*/
@Bean
public Exchange orderEventExchange() {
/*
* String name,
* boolean durable,
* boolean autoDelete,
* Map<String, Object> arguments
* */
return new TopicExchange("order-event-exchange", true, false);
}
@Bean
public Binding orderCreateBinding() {
/*
* String destination, 目的地(队列名或者交换机名字)
* DestinationType destinationType, 目的地类型(Queue、Exhcange)
* String exchange,
* String routingKey,
* Map<String, Object> arguments
* */
return new Binding("order.delay.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.create.order",
null);
}
@Bean
public Binding orderReleaseBinding() {
return new Binding("order.release.order.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.release.order",
null);
}
}
监听者
主类上添加注解
@RabbitListener(queues = "order.release.order.queue")
方法上添加注解
@RabbitHandler
public void listener(OrderEntity orderEntity, Channel channel, Message message)
只需要将参数设置号rabbitMQ就会自动传参
发送消息
rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",order.getOrder());
死性队列
安装
1. 首先我们将刚下载下来的 rabbitmq_delayed_message_exchange-3.8.0.ez文件上传到RabbitMQ所在服务器 2. 切换到插件所在目录, 执行 docker cp rabbitmq_delayed_message_exchange-3.8.0.ez rabbitmq:/plugins 命令,将刚插件拷贝到容器内plugins目录下 3. 执行 docker exec -it rabbitmq /bin/bash 命令进入到容器内部,并 cd plugins 进入plugins目录 4. 执行 ls -l|grep delay 命令查看插件是否copy成功 5. 在容器内plugins目录下,执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令启用插件 6. exit命令退出RabbitMQ容器内部,然后执行 docker restart rabbitmq 命令重启RabbitMQ容器
下载地址:Releases · rabbitmq/rabbitmq-delayed-message-exchange · GitHub
使用版本:
出现x-delayed-message交换机视为安装成功
更多推荐
所有评论(0)