Spring零配置下整合RabbitMQ（基础篇）

我不是女神ヾ 2022-05-31 05:30 177阅读 0赞

总共就配置类，生产者，消费者三个类和一个Properties配置文件。  
![这里写图片描述][SouthEast]

先看配置文件:

rabbit.username=guest
    rabbit.password=guest
    rabbit.host=localhost
    rabbit.port=5672
    rabbit.exchange=exchange
    rabbit.queue.default=defaultQueue
    rabbit.queue.service=serviceQueue
    rabbit.queue.default.key=defaultKey
    rabbit.queue.service.key=serviceKey

然后看生产者类:

import java.io.IOException;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    
    /** * @author wb-jjb318191 * @create 2018-02-08 9:58 */
    public class RabbitProducer { 
    
        private Logger logger = LoggerFactory.getLogger(RabbitProducer.class);
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Value("${rabbit.queue.default.key}")
        private String defaultKey;
    
        @Value("${rabbit.queue.service.key}")
        private String serviceKey;
    
        public void sendMessage(Object message) throws IOException {
            logger.info("发送Message:{}", message);
            rabbitTemplate.convertAndSend(defaultKey, message);
            rabbitTemplate.convertAndSend(serviceKey, message);
        }
    }

再看消费者类:

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    
    /** * Rabbit监听器 * * @author wb-jjb318191 * @create 2018-02-07 15:25 */
    public class RabbitConsumer { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RabbitConsumer.class);
    
        @RabbitListener(queues = "${rabbit.queue.default}")
        public void defaultConsumer(Message message) {
            LOGGER.info("DefaultConsumer消费消息:{}", message);
        }
    
        @RabbitListener(queues = "${rabbit.queue.service}")
        public void serviceConsumer(Message message) {
            LOGGER.info("ServiceConsumer消费消息:{}", message);
        }
    
    }

最后看配置类:

import com.bob.intergrate.rabbit.producer.RabbitProducer;
    import com.bob.intergrate.rabbit.consumer.RabbitConsumer;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Exchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.annotation.EnableRabbit;
    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.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.PropertySource;
    
    import static org.springframework.amqp.core.Binding.DestinationType.QUEUE;
    
    /** * RabbitMQ配置类 * * @author wb-jjb318191 * @create 2018-02-06 17:01 */
    @EnableRabbit
    @Configuration
    @PropertySource("classpath:rabbit-config.properties")
    public class RabbitContextConfig { 
    
        @Value("${rabbit.username}")
        private String username;
    
        @Value("${rabbit.password}")
        private String password;
    
        @Value("${rabbit.host}")
        private String host;
    
        @Value("${rabbit.port}")
        private int port;
    
        @Value("${rabbit.exchange}")
        private String exchange;
    
        @Value("${rabbit.queue.default}")
        private String defaultQueue;
    
        @Value("${rabbit.queue.default.key}")
        private String defaultKey;
    
        @Value("${rabbit.queue.service}")
        private String serviceQueue;
    
        @Value("${rabbit.queue.service.key}")
        private String serviceKey;
    
        @Bean
        public RabbitProducer rabbitProducer() {
            return new RabbitProducer();
        }
    
        @Bean
        public RabbitConsumer rabbitConsumer() {
            return new RabbitConsumer();
        }
    
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory factory = new CachingConnectionFactory();
            factory.setUsername(username);
            factory.setPassword(password);
            factory.setHost(host);
            factory.setPort(port);
            return factory;
        }
    
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
            return new RabbitAdmin(connectionFactory);
        }
    
        @Bean
        public Queue defaultQueue(RabbitAdmin rabbitAdmin) {
            Queue queue = new Queue(defaultQueue, true, false, false);
            queue.setAdminsThatShouldDeclare(rabbitAdmin);
            return queue;
        }
    
        @Bean
        public Queue serviceQueue(RabbitAdmin rabbitAdmin) {
            Queue queue = new Queue(serviceQueue, true, false, false);
            queue.setAdminsThatShouldDeclare(rabbitAdmin);
            return queue;
        }
    
        @Bean
        public Exchange directExchange(RabbitAdmin rabbitAdmin) {
            DirectExchange directExchange = new DirectExchange(exchange, true, false);
            directExchange.setAdminsThatShouldDeclare(rabbitAdmin);
            return directExchange;
        }
    
        @Bean
        public Binding defaultBinding(Exchange directExchange, Queue defaultQueue) {
            return new Binding(defaultQueue.getName(), QUEUE, directExchange.getName(), defaultKey, null);
        }
    
        @Bean
        public Binding serviceBinding(Exchange directExchange, Queue serviceQueue) {
            return new Binding(serviceQueue.getName(), QUEUE, directExchange.getName(), serviceKey, null);
        }
    
        @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Exchange directExchange) {
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            rabbitTemplate.setExchange(directExchange.getName());
            return rabbitTemplate;
        }
    
        @Bean
        public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
            containerFactory.setConnectionFactory(connectionFactory);
            return containerFactory;
        }
    
    }

之后就可以直接运行测试用例了:

@WebAppConfiguration
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(classes = RootContextConfig.class)
    public abstract class BaseControllerTest { 
    
        protected Gson gson;
    
        @Autowired
        protected WebApplicationContext webApplicationContext;
    
        @Before()
        public void setup() {
            //使用GsonBuilder针对日期类型指定解析后的格式,当Date只有年月日时,会使用当前时间来凑够解析长度
            gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
        }
    
    }
    
    @ContextConfiguration(classes = RabbitContextConfig.class)
    public class RabbitMQTest extends BaseControllerTest { 
    
        private Logger logger = LoggerFactory.getLogger(RabbitMQTest.class);
    
        @Test
        public void should_send_a_amq_message() throws Exception {
            RabbitProducer messageProducer = webApplicationContext.getBean("rabbitProducer", RabbitProducer.class);
            int a = 100;
            while (a > 0) {
                messageProducer.sendMessage("Hello, I am amq sender num :" + a--);
                try {
                    //暂停一下，好让消息消费者去取消息打印出来
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            }
        }
    }

基于纯Java形式的Spring和RabbitMQd的整合，只实现了最基础的流程，有了开头之后就会变得容易些了。

[SouthEast]: /images/20220531/868a00784056439b9d2717516f7e5ad9.png