RabbitMQ对接Spring Boot

短命女 2023-02-17 06:22 11阅读 0赞

# Spring AMQP #

Spring AMQP 是基于 Spring 框架的AMQP消息解决方案，提供模板化的发送和接收消息的抽象层，提供基于消息驱动的 POJO的消息监听等，很大方便我们使用RabbitMQ程序的相关开发。

Spring AMQP包含一些模块，如：spring-amqp, spring-rabbit and spring-erlang等，每个模块分别由独立的一些Jar包组成.

Spring AMQP模块主要包含org.springframework.amqp.core这个包中。这个包定义的相关类主要是与前面讲的AMQP模型相对应。Spring AMQP的目的是提供不依赖于任何特定的AMQP代理实现或客户端库通用的抽象。最终用户代码将很容易实现更易替换、添加和删除AMQP，因为它可以只针对抽象层来开发。这可以很方便我们选择和使用哪一个具体的broker实现，如sping-rabbit实现。

AMQP了解参考

LINK:[https://www.cnblogs.com/jmcui/p/8968315.html][https_www.cnblogs.com_jmcui_p_8968315.html]

LINK:[https://www.cnblogs.com/wade-luffy/p/6004219.html][https_www.cnblogs.com_wade-luffy_p_6004219.html]

## 1.使用RabbitAdmin ##

maven配置:

<dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

application.properties配置:

rabbiitmq.host =
    rabbiitmq.port = 
    rabbiitmq.username =
    rabbiitmq.password = 
    debug=true

编写RebbitMQConfig配置类:

@Value("${rabbiitmq.host}")
        private String host;
        @Value("${rabbiitmq.port}")
        private int port;
        @Value("${rabbiitmq.username}")
        private String username;
        @Value("${rabbiitmq.password}")
        private String password;
    
        /**
         * @Bean("name") 注入名称必须要和获取rabbitAdmin的入参名称一致(没有尝试过初始化方式)
         * @return
         */
        @Bean
        public ConnectionFactory connectionFactory(){
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
            connectionFactory.setHost(host);
            connectionFactory.setPort(port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost("/");
            return connectionFactory;
        }
    
        /**
         * 注入RabbitAdmin对象
         * @param connectionFactory 入参名称必须要和上面注入对象名称一致
         * @return
         */
        @Bean()
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
            RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
            //运行服务
            rabbitAdmin.setAutoStartup(true);
            return rabbitAdmin;
        }

使用方式:

@Autowired
        private RabbitAdmin rabbitAdmin;
    
        @Test
        public void testAdmin(){
            Map<String, Object> arguments = new HashMap<>();
            arguments.put("version","1.0");
            rabbitAdmin.declareExchange(new DirectExchange("test.direct",false,false));
            rabbitAdmin.declareExchange(new TopicExchange("test.topic",false,false));
            rabbitAdmin.declareExchange(new FanoutExchange("test.fanout",false,false));
            rabbitAdmin.declareExchange(new HeadersExchange("test.headers",false,false,arguments));
    
            rabbitAdmin.declareQueue(new Queue("test.direct.queue",false,false,false));
            rabbitAdmin.declareQueue(new Queue("test.topic.queue",false,false,false));
            rabbitAdmin.declareQueue(new Queue("test.fanout.queue",false,false,false));
            rabbitAdmin.declareQueue(new Queue("test.headers.queue",false,false,false,arguments));
    
            rabbitAdmin.declareBinding(new Binding(
                    "test.direct.queue",
                    Binding.DestinationType.QUEUE,
                    "test.direct",
                    "test.direct.#",null
            ));
            //第二种创建方式 链式创建方式-建立队列和交换机直接建立关系
            rabbitAdmin.declareBinding(
                    BindingBuilder.bind(
                            new Queue("test.direct.queue",
                                    false,
                                    false,
                                    false,
                                    null)
                    ).to(
                            new DirectExchange(
                                    "test.direct",
                                    false,
                                    false,
                                    null)
                    ).with("test.direct.#")
            );
            rabbitAdmin.declareBinding(new Binding(
                    "test.topic.queue",
                    Binding.DestinationType.QUEUE,
                    "test.topic",
                    "test.topic.#",null
            ));
            //第二种创建方式 链式创建方式-建立队列和交换机直接建立关系
            rabbitAdmin.declareBinding(
                    BindingBuilder.bind(
                            new Queue("test.topic.queue",
                                    false,
                                    false,
                                    false,
                                    null)
                    ).to(
                            new TopicExchange(
                                    "test.topic",
                                    false,
                                    false,
                                    null)
                    ).with("test.topic.#")
            );
    
            rabbitAdmin.declareBinding(new Binding(
                    "test.fanout.queue",
                    Binding.DestinationType.QUEUE,
                    "test.fanout",
                    "test.fanout.#",null
            ));
            //第二种创建方式 链式创建方式-建立队列和交换机直接建立关系 fanout不走路由key所以不需要with
            rabbitAdmin.declareBinding(
                    BindingBuilder.bind(
                            new Queue("test.fanout.queue",
                                    false,
                                    false,
                                    false,
                                    null)
                    ).to(
                            new FanoutExchange(
                                    "test.fanout",
                                    false,
                                    false,
                                    null)
                    )
            );
    
            rabbitAdmin.declareBinding(new Binding(
                    "test.headers.queue",
                    Binding.DestinationType.QUEUE,
                    "test.headers",
                    "test.headers.#",null
            ));
            //第二种创建方式 链式创建方式-建立队列和交换机直接建立关系 headers走得时附加参数所以写附加参数匹配规则
            rabbitAdmin.declareBinding(
                    BindingBuilder.bind(
                            new Queue("test.headers.queue",
                                    false,
                                    false,
                                    false,
                                    null)
                    ).to(
                            new HeadersExchange(
                                    "test.headers",
                                    false,
                                    false,
                                    null)
                    ).whereAll(arguments).match()
            );
            //清空队列
            rabbitAdmin.purgeQueue("test.topic.queue",false);
        }

或者使用注解的方式 将创建注入进来

private String queue001Name="queue001";
        private String topic001Name="topic001";
        private String queue002Name="queue002";
        private String topic002Name="topic002";
        /**
         * 可以直接用@Bean()的方式将exchange和queue以及binding直接注入到容器中RabbitmqAdmin会再启动时从容器中获取
         */
        @Bean
        public TopicExchange exchange001(){
            return new TopicExchange(topic001Name,true,false);
        }
        @Bean
        public Queue queue001(){
            return new Queue(queue001Name,false);
        }
        @Bean
        public Queue queue002(){
            return new Queue(queue002Name,false);
        }
        @Bean
        public Binding binding001() {
            String routingKey = "spring.*";
            //对象调用链
    //        return BindingBuilder.bind(queue001()).to(exchange001()).with(routingKey);
            return new Binding(
                    queue001Name,
                    Binding.DestinationType.QUEUE,
                    topic001Name,
                    routingKey, null
            );
        }

## 2.使用RabbitMQTemplate模板的方式发送消息 ##

在 RabbitMQAdmin 中新增 RabbitMQTemplat 配置:

//
        @Bean
        public RabbitTemplate getRabbitTemplate(ConnectionFactory connectionFactory){
            return new RabbitTemplate(connectionFactory);
        }

生产端:

@Test
        public void testTemplate2(){
            //1.创建消息和配置消息头
            MessageProperties messageProperties = new MessageProperties();
            //设置消息转换器识别的消息类型
            messageProperties.setContentType("text/plain");
            messageProperties.getHeaders().put("type1","测试1");
            messageProperties.getHeaders().put("type2","测试2");
            messageProperties.getHeaders().put("status","等待发送中");
            Message message = new Message("消息测试1".getBytes(),messageProperties);
            System.out.println(message.getMessageProperties().getHeaders().get("status"));
            rabbitTemplate.send("topic001", "spring.abcd1", message);
            rabbitTemplate.convertAndSend("topic001", "spring.abcd2", message, new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    //消息再发送后得处理
                    System.out.println("---------消息发送完毕--------");
                    System.out.println(new String(message.getBody()));
                    message.getMessageProperties().getHeaders().put("status","已发送");
                    System.out.println(message.getMessageProperties().getHeaders().get("status"));
                    return message;
                }
            });
        }

## 3.可以使用 SimpleMessageListenerContainer 消息监听容器: ##

在 RabbitMQTemplate  新增消息监听容器配置

/**
         * 使用 simplemessagelistenercontainer 消息监听容器
         * 监听队列和解析消息为对象
         * @param connectionFactory
         * @return
         */
        @Bean
        public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
            SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
            //监听多个队列
            simpleMessageListenerContainer.setQueues(queue001(),queue002());
            //设置消费者数量
            simpleMessageListenerContainer.setConcurrentConsumers(1);
            //设置最多并发消费者
            simpleMessageListenerContainer.setMaxConcurrentConsumers(5);
            //是否重回队列
            simpleMessageListenerContainer.setDefaultRequeueRejected(false);
            //设置签收模式
            simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
            //消费者标签(类似id)策略 用来反查消费者
            simpleMessageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
                @Override
                public String createConsumerTag(String queue) {
                    System.out.println("设置消费者标签");
                    return queue+"_"+ UUID.randomUUID().toString();
                }
            });
            //设置消费者
    /*        simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
                //单条消息消费
                @Override
                public void onMessage(Message message, Channel channel) throws Exception {
                    String messageStr = new String(message.getBody());
                    System.out.println("收到消息");
                    System.out.println(messageStr);
                }
            });*/
            //设置消息监听适配器
            MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(
                    new MessageDelegate()
            );
            //默认调用就是handleMessage(byte[] msg)方法
            messageListenerAdapter.setDefaultListenerMethod("handleMessage2");
            //将参数转换 可以转换为java对象或者java对象转字符串
            messageListenerAdapter.setMessageConverter(new TestMessageConverter());
            //消息监听容器设置监听适配器
            simpleMessageListenerContainer.setMessageListener(messageListenerAdapter);
            return simpleMessageListenerContainer;
    
        }

消费监听适配器:

package com.sy.rebbitmq_spring.beans;
    
    public class MessageDelegate {
        public void handleMessage(byte[] messageBody){
            System.out.println("默认方法:"+new String(messageBody));
        }
        public void handleMessage1(byte[] messageBody){
            System.out.println("默认方法1:"+new String(messageBody));
        }
        public void handleMessage2(String messageBody){
            System.out.println("默认方法2:"+messageBody);
        }
    }

消息转换器 可以将消息转换为java对象:

package com.sy.rebbitmq_spring.config;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.support.converter.MessageConversionException;
    import org.springframework.amqp.support.converter.MessageConverter;
    
    
    public class TestMessageConverter implements MessageConverter {
        //设置转换为消息
        @Override
        public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
            return new Message(object.toString().getBytes(),messageProperties);
        }
    
        //**设置转换为解析
        @Override
        public Object fromMessage(Message message) throws MessageConversionException {
            String contentType = message.getMessageProperties().getContentType();
            if (contentType != null && contentType.contains("text")){
                return new String(message.getBody());
            }
            return null;
        }
    }

##  具体代码仓库: ##

GITEE LINK: [https://gitee.com/suyibk/rebbitmq\_spring][https_gitee.com_suyibk_rebbitmq_spring]

[https_www.cnblogs.com_jmcui_p_8968315.html]: https://www.cnblogs.com/jmcui/p/8968315.html
[https_www.cnblogs.com_wade-luffy_p_6004219.html]: https://www.cnblogs.com/wade-luffy/p/6004219.html
[https_gitee.com_suyibk_rebbitmq_spring]: https://gitee.com/suyibk/rebbitmq_spring