springboot 集成 rabbitmq 实战应用

springboot 集成 rabbitmq 实战应用今天分享：分为生产者和消费者两个角色

一、生产者

1、pom文件引入

        
            org.springframework.boot
            spring-boot-starter-amqp

2、yml配置文件添加

spring:
  rabbitmq:
    host: 194.11.33.120
    port: 5672
    username: admin
    password: admin
    publisher-confirms: true
    publisher-returns: true
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual

3、对象的初始化


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;


@Slf4j
@Configuration
public class RabbitConfig {

    public final static String EXCHANGE_LOG = "order.log.producer.reply";
    public final static String KEY_LOG = "order.log.reply";

    @Value("${spring.rabbitmq.host}")
    private String addresses;

    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirms;

    @Value("${spring.rabbitmq.publisher-returns}")
    private boolean publisherReturns;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses+":"+port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        
//        connectionFactory.setPublisherConfirms(publisherConfirms);
//        connectionFactory.setPublisherReturns(publisherReturns);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        return new RabbitAdmin(connectionFactory);
    }


    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, List list) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(2);
//抓取参数非常关键，一次抓取的消息多了，消费速度一慢，就会造成响应延迟，抓取少了又会导致并发量低，消息堵塞
        factory.setPrefetchCount(10);

        
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        return factory;
    }

    @Bean
    public Queue queuelogMessage() {
        return new Queue("order.log.queue.reply");
    }

    @Bean
    public DirectExchange exchange() {
        return new DirectExchange(EXCHANGE_LOG);
    }

    @Bean
    public Binding bindingLogExchangeMessage() {
        return BindingBuilder
                .bind(queuelogMessage())
                .to(exchange())
                .with(KEY_LOG);
    }
}

4、发送信息的工具类

 

import cn.nan.config.RabbitConfig;
import cn.nan.mall.model.MessageLog;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class OrderLogSender {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String actionId) {
        log.info("TopicSender send the 1st : " + actionId);
        this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_LOG, RabbitConfig.KEY_LOG, actionId);
    }

    public void send(MessageLog messageLog, CorrelationData correlationData) {
        String msg1 = JSON.toJSONString(messageLog);
        log.info("TopicSender send the 1st : " + msg1);
        this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_LOG, RabbitConfig.KEY_LOG, msg1,correlationData);
    }
}

二、消费者

1、pom和yml配置同生产者一致

2、对象实例化:里面有消费者的配置，同时也有生产者的配置

import cn.nan.mall.model.MessageLog;
import cn.nan.mall.service.MessageLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;


@Slf4j
@Configuration
public class RabbitConfig {

    public final static String EXCHANGE_LOG = "order.log.producer";
    public final static String KEY_LOG = "order.log";

    @Value("${spring.rabbitmq.host}")
    private String addresses;

    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirms;

    @Value("${spring.rabbitmq.publisher-returns}")
    private boolean publisherReturns;

    @Autowired
    private MessageLogService messageLogService;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses+":"+port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        
        connectionFactory.setPublisherConfirms(publisherConfirms);
        connectionFactory.setPublisherReturns(publisherReturns);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate newRabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息成功发送到Exchange,messageId:" + correlationData.getId());
                //修改日志表状态，状态改成 1，投递成功且未确认
                if(updateMessageLog(1,Long.valueOf(correlationData.getId())) == 1) {
                    log.info("------modify status 1 ok--------");
                }
            } else {
                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
            }
        });
        template.setMandatory(true);
        template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
        });
        //不使用临时队列
//        template.setUseTemporaryReplyQueues(false);
//        template.setReplyAddress("amq.rabbitmq.reply-to");
//        template.setUserCorrelationId(true);
//        template.setReplyTimeout(10000);
        return template;
    }

    private int updateMessageLog(int status,Long messageId) {
        MessageLog messageLog = new MessageLog();
        messageLog.setMessageId(messageId);
        messageLog.setStatus(status);
        messageLog.setTryCount(1);
        return messageLogService.updateMessageLog(messageLog);
    }


    @Bean
    public Queue queuelogMessage() {
        return new Queue("order.log.queue");
    }

    @Bean
    public DirectExchange exchange() {
        return new DirectExchange(EXCHANGE_LOG);
    }

    @Bean
    public Binding bindingLogExchangeMessage() {
        return BindingBuilder
                .bind(queuelogMessage())
                .to(exchange())
                .with(KEY_LOG);
    }

    
    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, List list) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(2);
//抓取参数非常关键，一次抓取的消息多了，消费速度一慢，就会造成响应延迟，抓取少了又会导致并发量低，消息堵塞
        factory.setPrefetchCount(10);

        
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        return factory;
    }

    //===============生产者发送确认==========
    @Bean
    public RabbitTemplate.ConfirmCallback confirmCallback(){
        return new RabbitTemplate.ConfirmCallback(){

            @Override
            public void confirm(CorrelationData correlationData,
                                boolean ack, String cause) {
                if (ack) {
                    log.info("发送者确认发送给mq成功");
                } else {
                    //处理失败的消息
                    log.info("发送者发送给mq失败,考虑重发:"+cause);
                }
            }
        };
    }

    @Bean
    public RabbitTemplate.ReturnCallback returnCallback(){
        return new RabbitTemplate.ReturnCallback(){

            @Override
            public void returnedMessage(Message message,
                                        int replyCode,
                                        String replyText,
                                        String exchange,
                                        String routingKey) {
                log.info("无法路由的消息，需要考虑另外处理。");
                log.info("Returned replyText："+replyText);
                log.info("Returned exchange："+exchange);
                log.info("Returned routingKey："+routingKey);
                String msgJson  = new String(message.getBody());
                log.info("Returned Message："+msgJson);
            }
        };
    }
}

3、监听消费

 

import cn.nan.mall.model.MessageLog;
import cn.nan.mall.service.MessageLogService;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.util.Map;


@Slf4j
@Component
public class OrderLogReceiver {

    @Autowired
    private MessageLogService messageLogService;

    @RabbitListener(queues = "order.log.queue.reply")
    @RabbitHandler // 此注解加上之后可以接受对象型消息
    public void onMessage(Message message, Channel channel,@Headers Map headers) throws Exception {
        try {
            String msg = new String(message.getBody());
            log.info("OrderLogReceiver>>>>>>>message received:" + msg);
            MessageLog messageLog = JSON.parseObject(msg, MessageLog.class);
            try {
                messageLog.setStatus(9);
                messageLog.setTryCount(null);
                messageLogService.updateMessageLog(messageLog);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手工确认，可接下一条
            } catch (Exception e) {
                e.printStackTrace();
                log.info(e.getMessage());
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);//失败，则直接忽略此订单
                log.info("OrderLogReceiver>>>>>>message nack");
                throw e;
            }
        } catch (Exception e) {
            log.info(e.getMessage());
        }
    }
}

deliveryTag（唯一标识 ID）：当一个消费者向 RabbitMQ 注册后，会建立起一个 Channel ，Rab bitMQ 会用 basic.deliver 方法向消费者推送消息，这个方法携带了一个 delivery tag，
它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，delivery tag 的范围仅限于 Channel
multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息

到此、springboot 集成 rabbitmq 实战应用分享完毕，下篇会分享一些细节，顺序、重复消费、消息丢失等等，敬请期待！

springboot 集成 rabbitmq 实战应用

Java相关栏目本月热门文章