Springboot 集成apollo消息中间件

我就是我 2022-04-17 07:16 451阅读 0赞

SpringBoot集成appllo，先看效果：

生成消息的代码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70][]

效果图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70 1][]

消费消息的代码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70 2][]

本人是通过camel进行消息推送和接收的。

1） 引入jar包：

<dependency>
    			<groupId>org.apache.camel</groupId>
    			<artifactId>camel-spring-boot-starter</artifactId>
    			<version>2.17.3</version>
    </dependency>
    <dependency>
    			<groupId>org.apache.camel</groupId>
    			<artifactId>camel-core</artifactId>
    			<version>2.17.0</version>
    </dependency>
    <dependency>
    			<groupId>org.apache.activemq</groupId>
    			<artifactId>activemq-camel</artifactId>
    			<version>5.9.0</version>
    </dependency>

2）配置apollo：

server.port=8077
    
    # apollo连接
    apollo.url=tcp://127.0.0.1:61613
    apollo.password=password
    apollo.user=admin
    
    
    apollo.queue=activemq://queue
    apollo.queue.consumer=activemq://queue.command.response

package com.apollo.config;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.pool.PooledConnectionFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author :zoboy
     * @Description:
     * @ Date: Created in 2018-11-14 15:00
     */
    @Configuration
    public class ActivemqCamelAutoConfiguration {
    
        @Value("${apollo.url}")
        private String url;
        @Value("${apollo.password}")
        private String psw;
        @Value("${apollo.user}")
        private String user;
    
    
        @Bean
        PooledConnectionFactory mqConnectionFactory(){
    
            PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(new ActiveMQConnectionFactory(user,psw,url));
    
            return pooledConnectionFactory;
        }
    
    }

3）生产者:

package com.apollo.produce;
    
    import org.apache.camel.ProducerTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @author :zoboy
     * @Description:
     * @ Date: Created in 2018-11-14 15:03
     */
    @RestController
    public class ProduceController {
    
        // activemq://queue
        @Value("${apollo.queue}")
        private String queue;
        @Autowired
        private ProducerTemplate producerTemplate;
    
        @GetMapping("/send/{jsonStr}")
        public void send(@PathVariable String jsonStr){
            producerTemplate.sendBody(queue,jsonStr);
        }
    }

4)消费者：

这儿需要camel路由一下：

package com.apollo.route;
    
    import com.apollo.consumer.ConsumerHandle;
    import org.apache.camel.builder.RouteBuilder;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    /**
     * @author :zoboy
     * @Description:
     * @ Date: Created in 2018-11-14 15:23
     */
    @Component
    public class MtqqMessageRouter extends RouteBuilder {
        @Autowired
        ConsumerHandle consumerHandle;
    
        @Value("${apollo.queue.consumer}")
        private String queue;
    
        @Override
        public void configure() throws Exception {
            from(queue).streamCaching().process(consumerHandle);
        }
    }

消费：

package com.apollo.consumer;
    
    import org.apache.camel.Exchange;
    import org.apache.camel.Message;
    import org.apache.camel.Processor;
    import org.springframework.stereotype.Component;
    
    /**
     * @author :zoboy
     * @Description:
     * @ Date: Created in 2018-11-14 15:22
     */
    @Component
    public class ConsumerHandle implements Processor {
    
        private static final ThreadLocal<Exchange> localExchange = new ThreadLocal<Exchange>();
        @Override
        public void process(Exchange exchange) throws Exception {
            setLocalExchange(exchange);
    
            try{
                doProcess(exchange.getIn());
            }finally{
                removeLocalExchange();
            }
        }
        public void setLocalExchange(Exchange exchange) {
            localExchange.set(exchange);
        }
    
        public void removeLocalExchange() {
            localExchange.remove();
        }
    
        public  void doProcess(Message message) throws Exception{
            // message 就是发送方获取的消息
            System.out.println(message.getBody().toString());
        }
    }

代码完毕：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70 3][]

项目完整地址：

[https://github.com/hzoboy/springboot\_apollo.git][https_github.com_hzoboy_springboot_apollo.git]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70]: /images/20220417/ef66e59d84964bf59bfbc266bc681c23.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70 1]: /images/20220417/9ca1111e19f84dbd947f771b46dfc5af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70 2]: /images/20220417/ab5baf07943c4b1080a73a41624b0b24.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEwNTE5MTI_size_16_color_FFFFFF_t_70 3]: /images/20220417/235323434a3842f5ad6588cbaa9828b6.png
[https_github.com_hzoboy_springboot_apollo.git]: https://github.com/hzoboy/springboot_apollo.git