流式消息_spring-cloud-starter-stream-rocketmq发送消息_v1.0.1

阳光穿透心脏的1/2处 2023-01-09 12:52 97阅读 0赞

<table> 
 <thead> 
  <tr> 
   <th>文件名称</th> 
   <th>版本号</th> 
   <th>作者</th> 
   <th>qq</th> 
   <th>组件版本</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>流式消息_spring-cloud-starter-stream-rocketmq发送消息</td> 
   <td>v1.0.1</td> 
   <td>学生宫布</td> 
   <td>8416837</td> 
   <td>SpringBoot 2.1.8.RELEASE <br>spring-cloud-starter-stream-rocketmq 2.1.1.RELEASE</td> 
  </tr> 
 </tbody> 
</table>

### 文章目录 ###

*   *  简介
     *   *  说明
     *  模式
     *   *  发布
         *   *  代码
         *  订阅 同理 不过代码有不同
         *   *  代码

## 简介 ##

### 说明 ###

通道式发布|订阅，如果更换消息中间件技术栈，勿须改动代码。

## 模式 ##

### 发布 ###

#### 代码 ####

1.  引入依赖，以及其它相关类，当前假设已经有完整项目

<dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            </dependency>

1.  在配置文件配置，或yaml文件：

spring.cloud.stream.rocketmq.binder.name-server=ipv4:9876
    # 相当于topic
    spring.cloud.stream.bindings.output.destination=zhangsan_msg
    spring.cloud.stream.bindings.output.group=zhangsan_msg_group_001
    # 订阅
    # 相当于topic
    # spring.cloud.stream.bindings.input.destination=zhangsan_msg
    # spring.cloud.stream.bindings.input.group=lisi-group-consumer_9527

1.  通道类

package org.springframework.cloud.stream.messaging;
    
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    public interface Source { 
    	// output与配置对应 ...bindings.output...
        String OUTPUT = "output";
    
        @Output(OUTPUT)
        MessageChannel output();
    }

1.  启动类加注解，也可加载在业务类上面

// 可以绑定多个
    @EnableBinding({ Source.class, Xxx.class, Xyy.class})

1.  生产代码，【注意】下文output与配置文件、interface文件里的output名称保持一样：

@Resource
        private MessageChannel output;
        ... ....
        // 函数内
        ret = output.send(MessageBuilder.withPayload("cc-使用通道-MessageChannel模式发送消息".concat(msg)).build());

1.  发送成功，在ng控制台查看：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NjMDA3Y2MwMDk_size_16_color_FFFFFF_t_70]

### 订阅 同理 不过代码有不同 ###

#### 代码 ####

1.  配置

# 订阅
    # 相当于topic
    spring.cloud.stream.bindings.input.destination=zhangsan_msg
    spring.cloud.stream.bindings.input.group=lisi-group-consumer_9527
    
    spring.cloud.stream.bindings.ccWriteInput.destination=cc_msg
    spring.cloud.stream.bindings.ccWriteInput.group=user-group-consumer_9528

1.  通道类，【提示】CCInputOutput.java可以写多个output、input，注意名称对应即可：

public interface CCInputOutput { 
        String OUTPUT_CC_WRITE = "ccWriteOutput";
    
        @Output(OUTPUT_CC_WRITE)
        MessageChannel ccWriteOutput();
    
        String INPUT_CC_WRITE = "ccWriteInput";
    
        @Input(INPUT_CC_WRITE)
        SubscribableChannel ccWriteInput();
    }

1.  生产和消费代码

> 生产

@Resource
        MessageChannel ccWriteOutput;
        ... ...
    	    User u1 = new User();
            u1.setIdNumber("123789");
            u1.setFullName("罗士信");
            u1.setAddress("河西走廊");
            boolean ret = ccWriteOutput.send(MessageBuilder.withPayload(u1).build());

> 消费

// 上文的INPUT_CC_WRITE
    	@StreamListener(CCInputOutput.INPUT_CC_WRITE)
        public void receive(String messageBody){ 
    
            log.debug("通过stream收到消息，messageBody:{}", messageBody);
            // 处理messageBody 略
            ... ...

1.  成功收到消息：  
    ![在这里插入图片描述][20210121143109318.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NjMDA3Y2MwMDk_size_16_color_FFFFFF_t_70]: /images/20221119/ce2fa71372264538947ace3a478983b9.png
[20210121143109318.png]: /images/20221119/f2cdf05e8da14e579c7ea9c4e5b39040.png