springboot+mq
springboot整合rabbitmq,支持消息确认机制
安装
推荐一篇博客 https://blog.csdn.net/zhuzhezhuzhe1/article/details/80464291
项目结构
image
POM.XML
image
POM.XML
application.yml
需要将publisher-confrems设为true，启动确认回调, 将 publisher-returns设为true 确认返回回调
image
rabbitmq配置类--RabbitConfig
第一部分, 定义队列
image
第二部分，设置一些消息处理策略
image
/**
- rabbitMq 配置类
- @author milicool
- Created on 2018/9/14 /
@Configuration
public class RabbitConfig {
@Resource
private RabbitTemplate rabbitTemplate;
/* - 定义一个hello的队列
- Queue 可以有4个参数
1.队列名2.durable 持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true3.auto-delete 表示消息队列没有在使用时将被自动删除 默认是false4.exclusive 表示该消息队列是否只在当前connection生效,默认是false */
@Bean
public Queue helloQueue() {
return new Queue("queue-test");
}
/** ======================== 定制一些处理策=============================*/
- ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调 即消息发送到exchange ack
- ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调 即消息发送不到任何一个队列中 ack */
@Bean public RabbitTemplate rabbitTemplate() {
Logger log = LoggerFactory.getLogger(RabbitTemplate.class);
// 消息发送失败返回到队列中, yml需要配置 publisher-returns: true
rabbitTemplate.setMandatory(true);
// 消息返回, yml需要配置 publisher-returns: true
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { String correlationId = message.getMessageProperties().getCorrelationIdString(); log.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {} 路由键: {}", correlationId, replyCode, replyText, exchange, routingKey); });
// 消息确认, yml需要配置 publisher-confirms: true
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if (ack) { // log.debug("消息发送到exchange成功,id: {}", correlationData.getId());
} else { log.debug("消息发送到exchange失败,原因: {}", cause); } });
return rabbitTemplate; } }
生产者
/**
- 生产者
- @author milicool
- Created on 2018/9/14
/
@Component
public class Producer {
@Autowired private RabbitTemplate rabbitTemplate;
/* - 给hello队列发送消息 */
public void send() { for (int i =0; i< 100; i++) { String msg = "hello, 序号: " + i; System.out.println("Producer, " + msg); rabbitTemplate.convertAndSend("queue-test", msg); } }
}
消费者
/**
-
消费者
-
@author milicool
-
Created on 2018/9/14
*/
@Component
public class Comsumer { private Logger log = LoggerFactory.getLogger(Comsumer.class);@RabbitListener(queues = "queue-test")
public void process(Message message, Channel channel) throws IOException {
// 采用手动应答模式, 手动确认应答更为安全稳定
channel.basicAck(message.getMessageProperties().getDeliveryTag(), true); log.info("receive: " + new String(message.getBody())); } }
测试类
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {
@Autowired
private Producer producer;
@Test
public void contextLoads() { producer.send(); }
}</pre>
复制代码
](javascript:void(0); "复制代码")
测试结果
测试结果太长，没有截取全部，可以查看到消费者接收到了全部消息，如果有的消息在没有接收完，消息将被持久化，下次启动时消费
image
web端查看
image
