Spring Boot集成RabbitMQ

谁践踏了优雅 2022-04-18 05:48 187阅读 0赞

## 订阅程序 ##

*  源码

@Bean
    public SimpleMessageListenerContainer customerMessageContainer() { 
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(orderConnectionFactory());
        container.setQueues(recoveryQueue());
        container.setExposeListenerChannel(true);
        container.setPrefetchCount(1);// 每次只消费1条
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 设置手动应答
        container.setConcurrentConsumers(1); // Specify the number of concurrent consumers to create. Default is 1.
        container.setMessageListener(new ChannelAwareMessageListener() { 
            public void onMessage(Message message, Channel channel) throws Exception { 
                byte[] body = message.getBody();
                String bodyStr = new String(body);
                try { 
                    LOGGER.info("正在推送99单号回收汇总信息");
                    dailyRecoveryGatherService.pushDailyRecoveryGather(bodyStr);
                    // 确认应答（是否批量.true:将一次性ack所有小于deliveryTag的消息）
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    LOGGER.info("99单号回收汇总推送完成,消息内容:" + bodyStr);
                } catch (Exception e) { 
                    LOGGER.error("99单号回收汇总推送失败,消息内容:" + bodyStr, e);
                    /** * 拒绝应答（true表示重新入队列， * false表示直接从队列中删除，此时和basicAck(long deliveryTag, false)的效果一样） */
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                }
            }
        });
        return container;
    }

## 创建管道和队列日志 ##

org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:360) 2019-01-04 09:02:25 [INFO] Created new connection: orderConnectionFactory#368a8022:0/SimpleConnection@ef99e17 [delegate=amqp://guest@192.168.0.204:5672/, localPort= 50346]
    org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:360) 2019-01-04 09:02:25 [INFO] Created new connection: SpringAMQP#38d84b3:0/SimpleConnection@19a04ff3 [delegate=amqp://guest@192.168.0.204:5672/, localPort= 50347]
    org.springframework.boot.StartupInfoLogger.logStarted(StartupInfoLogger.java:57) 2019-01-04 09:02:25 [INFO] Started Application in 3.334 seconds (JVM running for 3.948)

## RabbitMQ 可靠投递 ##

*  大神博客-[RabbitMQ 可靠投递][RabbitMQ]
 *  [rabbitmq使用总结][rabbitmq]

[RabbitMQ]: https://www.cnblogs.com/wangiqngpei557/p/9381478.html
[rabbitmq]: https://www.cnblogs.com/frankltf/p/10373524.html