spring-boot整合rabbitMQ

旧城等待， 2022-01-14 12:49 311阅读 0赞

MQ配置项

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitmqConfig {
    
    
        @Autowired
        private CachingConnectionFactory connectionFactory;
    
        @Autowired
        private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;
    
        private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);
    
        @Bean
        public RabbitTemplate rabbitTemplate() {
            connectionFactory.setPublisherConfirms(true);
            connectionFactory.setPublisherReturns(true);
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            rabbitTemplate.setMandatory(true);
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause));
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message));
            return rabbitTemplate;
        }
    
        @Bean(name = "justiceQueue")
        public Queue justiceQueue() {
            return new Queue("justiceQueue", true);
        }
    
        @Bean
        public DirectExchange justiceExchange() {
            return new DirectExchange("justiceExchange", true, false);
        }
    
        @Bean
        public Binding justiceBinding(Queue queue, DirectExchange directExchange) {
            return BindingBuilder.bind(queue).to(directExchange).with("justiceRoutingKey");
        }
    }

MQ测试controller

import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import ojama.entity.Mail;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageBuilder;
    import org.springframework.amqp.core.MessageDeliveryMode;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    @RestController
    public class TestController {
        @Resource
        private RabbitTemplate rabbitTemplate;
        @Resource
        private ObjectMapper objectMapper;
    
        @RequestMapping("/hello")
        public Mail userRegister(String name, String password) throws JsonProcessingException {
            Mail mail = new Mail()
                    .setContent("hello!,your name is " + name + ",your password is " + password)
                    .setFrom("785607860@qq.com")
                    .setTo("1061037698@qq.com")
                    .setSubject("hello");
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.setExchange("justiceExchange");
            rabbitTemplate.setRoutingKey("justiceRoutingKey");
            Message message = MessageBuilder.withBody(objectMapper.writeValueAsBytes(mail)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            message.getMessageProperties().setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME, MessageProperties.CONTENT_TYPE_JSON);
            rabbitTemplate.convertAndSend(message);
            return mail;
        }
    }

Mail实体

import java.io.Serializable;
    
    public class Mail implements Serializable {
        private String from;
        private String to;
        private String subject;
        private String content;
    
        public String getFrom() {
            return from;
        }
    
        public Mail setFrom(String from) {
            this.from = from;
            return this;
        }
    
        public String getTo() {
            return to;
        }
    
        public Mail setTo(String to) {
            this.to = to;
            return this;
        }
    
        public String getSubject() {
            return subject;
        }
    
        public Mail setSubject(String subject) {
            this.subject = subject;
            return this;
        }
    
        public String getContent() {
            return content;
        }
    
        public Mail setContent(String content) {
            this.content = content;
            return this;
        }
    }

消息接收类:

import com.fasterxml.jackson.databind.ObjectMapper;
    import ojama.entity.Mail;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    @Component
    @RabbitListener(queues = "justiceQueue")
    public class MqHandler {
    
        @Resource
        private ObjectMapper objectMapper;
    
        private static final Logger log = LoggerFactory.getLogger(MqHandler.class);
    
        @RabbitHandler
        public void forTheKing(@Payload byte[] message) {
            try {
                Mail mail = objectMapper.readValue(message, Mail.class);
                System.out.println(mail);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

application.yml配置

spring:
      rabbitmq:
        host: localhost
        username: mqadmin
        password: mqadminpassword

参考文章:[https://blog.csdn.net/u013871100/article/details/82982235][https_blog.csdn.net_u013871100_article_details_82982235]

[https_blog.csdn.net_u013871100_article_details_82982235]: https://blog.csdn.net/u013871100/article/details/82982235