SpringCloud alibaba 2.1.X (八) Stream-RocketMQ 消息生产消费

分手后的思念是犯贱 2023-07-03 04:46 19阅读 0赞

# 1.安装RocketMQ #

请参考 : [docker 安装 rocketmq(namesrv broker 与可视化控制台 console)][docker _ rocketmq_namesrv broker _ console]

# 2.项目父工程 #

请参考 : [SpringCloud H系列 alibaba 2.1.1 (一) nacos1.X 注册中心 服务端][SpringCloud H_ alibaba 2.1.1 _ nacos1.X _]

# 3.项目结构  #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70][]

# 4.创建消息生产端 alibaba-stream-rocketmq-produce #

pom.xml

<parent>
            <groupId>com.alibaba</groupId>
            <artifactId>spring-cloud-hoxton</artifactId>
            <version>1.0</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>alibaba-stream-rocketmq-produce</artifactId>
        <packaging>jar</packaging>
    
        <dependencies>
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
        </dependencies>

application.yml  配置文件

application:
        name: alibaba-stream-rocketmq-produce
      cloud:
        stream:
          rocketmq:
            binder:
              # rocketmq 地址
              name-server: 192.168.101.11:9876
          bindings:
            output:
              content-type: application/json
              # 用来指定topic
              destination: stream-test-topic

StreamRocketMQProduceApplication.java  启动类  增加 @EnableBinding(Source.class) 注解

/**
     * @author Lion Li
     */
    @SpringBootApplication
    @EnableBinding(Source.class)
    public class StreamRocketMQProduceApplication {
    
    	public static void main(String[] args) {
    		SpringApplication.run(StreamRocketMQProduceApplication.class, args);
    	}
    
    }

创建消息体  TestMessaging.java 这里使用了lombok

/**
     * @author Lion Li
     */
    @Data
    @Accessors(chain = true)
    public class TestMessaging {
        /**
         * 消息id
         */
        private String msgId;
        /**
         * 消息内容
         */
        private String msgText;
    }

创建消息投递方法 TestController.java

/**
     * @author Lion Li
     */
    @RestController
    public class TestController {
    
        @Autowired
        private Source source;
    
        @GetMapping("/stream-test-msg")
        public String streamTestMsg(String msg){
            // 构建消息对象
            TestMessaging testMessaging = new TestMessaging()
                    .setMsgId(UUID.randomUUID().toString())
                    .setMsgText(msg);
            source.output().send(
                    MessageBuilder.withPayload(testMessaging).build());
            return "投递消息 => " + msg + " => 成功";
        }
    }

启动消息生产端  访问 http://localhost:8000/stream-test-msg?msg=Lion Li

![20200202123406114.png][]

查看rocketmq控制台

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 1][]

查看消息内容

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 2][]

投递成功

# 5.创建消息生产端 alibaba-stream-rocketmq-consumer #

pom.xml

<parent>
            <groupId>com.alibaba</groupId>
            <artifactId>spring-cloud-hoxton</artifactId>
            <version>1.0</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>alibaba-stream-rocketmq-consumer</artifactId>
        <packaging>jar</packaging>
    
        <dependencies>
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
        </dependencies>

application.yml 配置文件

server:
      port: 8002
    spring:
      application:
        name: alibaba-stream-rocketmq-consumer
      cloud:
        stream:
          rocketmq:
            binder:
              # rocketmq 地址
              name-server: 192.168.101.11:9876
          bindings:
            input:
              content-type: application/json
              # 定义 topic
              destination: stream-test-topic
              # group一定要写
              group: test-group

StreamRocketMQConsumerApplication.java  启动类  增加 @EnableBinding(Sink.class) 注解

/**
     * @author Lion Li
     */
    @SpringBootApplication
    @EnableBinding(Sink.class)
    public class StreamRocketMQConsumerApplication {
    
    	public static void main(String[] args) {
    		SpringApplication.run(StreamRocketMQConsumerApplication.class, args);
    	}
    
    }

消息体  TestMessaging.java

创建消息监听  TestStreamConsumer.java  包含全局异常处理

@Slf4j
    @Service
    public class TestStreamConsumer {
    
        @StreamListener(Sink.INPUT)
        public void receive(TestMessaging testMessaging){
            log.warn("通过stream消费到消息 => {}",testMessaging.toString());
    //        if (testMessaging.getMsgText().equals("测试异常")){
    //            throw new RuntimeException("测试全局异常拦截");
    //        }
        }
    
        /**
         * 全局异常处理
         */
        @StreamListener("errorChannel")
        public void error(Message<?> message){
            ErrorMessage errorMessage = (ErrorMessage) message;
            log.error("发生异常 => {}",errorMessage);
        }
    
    }

启动消息消费端  查看是否可以消费到消息

![20200202124230223.png][]

消费成功

项目已上传到gitee

地址: [spring-cloud-alibaba-H-demo][]

如果帮到您了,请帮忙点个star

[docker _ rocketmq_namesrv broker _ console]: https://blog.csdn.net/weixin_40461281/article/details/104115052
[SpringCloud H_ alibaba 2.1.1 _ nacos1.X _]: https://blog.csdn.net/weixin_40461281/article/details/103384273
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200202122249385.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ==,size_16,color_FFFFFF,t_70
[20200202123406114.png]: https://img-blog.csdnimg.cn/20200202123406114.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200202123448535.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200202123500441.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ==,size_16,color_FFFFFF,t_70
[20200202124230223.png]: https://img-blog.csdnimg.cn/20200202124230223.png
[spring-cloud-alibaba-H-demo]: https://gitee.com/JavaLionLi/spring-cloud-alibaba-H-demo