springboot+stream-rabbit学习笔记

博主： hdy
发布时间：2019 年 07 月 02 日
3989 次浏览
暂无评论
8252字数
分类：架构后端

引入pom.xml

        <!-- https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-stream-rabbit -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
            <version>2.1.3.RELEASE</version>
        </dependency>

配置client

/**
 * @author huangdeyao
 * @date 2019/7/2 12:50
 */
public interface StreamClient {
 /**
  * 同一个服务里面的信道名字不能一样，在不同的服务里可以相同名字的信道
  */
 String STREAM_INPUT = "stream-input";
 String STREAM_OUTPUT = "stream-output";

 /**
  * 订阅消息
  *
  * @return
  */
 @Input(StreamClient.STREAM_INPUT)
 SubscribableChannel input();

 /**
  * 消息发送
  *
  * @return
  */
 @Output(StreamClient.STREAM_OUTPUT)
 MessageChannel output();
}

接收端

/**
 * @author huangdeyao
 * @date 2019/7/2 13:04
 */
@Component
@EnableBinding(StreamClient.class)
@Log4j2
public class StreamReceiver {

 @StreamListener(StreamClient.STREAM_INPUT)
 public void receive(Object message) {
     log.info("stream input receiver: {}", message);
 }
}

发送端

 @Autowired
 StreamClient streamClient;

 @GetMapping("/send")
 public String messageWithMQ(String message) {
     streamClient.output().send(MessageBuilder.withPayload(message).build());
     return "ok";
 }

遇到的错误

org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'stream_input' defined in com.dy.spring.client.StreamClient: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.dy.spring.client.StreamClient; factoryMethodName=input; initMethodName=null; destroyMethodName=null
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:68) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerOutputBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:55) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.lambda$registerBindingTargetBeanDefinitions$0(BindingBeanDefinitionRegistryUtils.java:92) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:410) ~[spring-core-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:389) ~[spring-core-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(BindingBeanDefinitionRegistryUtils.java:82) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.cloud.stream.config.BindingBeansRegistrar.registerBeanDefinitions(BindingBeansRegistrar.java:44) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.lambda$loadBeanDefinitionsFromRegistrars$1(ConfigurationClassBeanDefinitionReader.java:364) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[na:1.8.0_191]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:363) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:145) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:117) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:327) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:232) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:275) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:95) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:705) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:531) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:742) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:389) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.run(SpringApplication.java:311) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.run(SpringApplication.java:1213) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.run(SpringApplication.java:1202) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at com.dy.spring.SpringDomeApplication.main(SpringDomeApplication.java:13) [classes/:na]

Disconnected from the target VM, address: '127.0.0.1:57769', transport: 'socket'

Process finished with exit code 1

解决StreamClient 中不能通道名，使用不同的名字
修改

public interface StreamClient {
 /**
  * 同一个服务里面的信道名字不能一样，在不同的服务里可以相同名字的信道
  */
 String STREAM_INPUT = "stream-input";
 String STREAM_OUTPUT = "stream-output";
 
...... 略
}
yml配置,互相指向成一个通道stream-output
spring:
  cloud:
 stream:
   bindings:
     stream-input:
       destination: stream-output

多实例接收多个消息解决，yml配置group

spring:
  rabbitmq:
 port: 5672
 addresses: 192.168.177.129
 username: guest
 password: guest
  cloud:
 stream:
   bindings:
     stream-input:
       destination: stream-output
       group: my-stream

接收后回调信息

StreamClient.class

    /**
     * 同一个服务里面的信道名字不能一样，在不同的服务里可以相同名字的信道
     */
    String STREAM_INPUT_2 = "stream-input-2";
    String STREAM_OUTPUT_2 = "stream-output-2";
  /**
     * 订阅消息
     *
     * @return
     */
    @Input(StreamClient.STREAM_INPUT_2)
    SubscribableChannel input2();

    /**
     * 消息发送
     *
     * @return
     */
    @Output(StreamClient.STREAM_OUTPUT_2)
    MessageChannel output2();

StreamReceiver.class

    @StreamListener(StreamClient.STREAM_INPUT)
    @SendTo(StreamClient.STREAM_INPUT_2)
    public UserEntity process(UserEntity userEntity) {
        log.info("stream input receiver: {}", userEntity.toString());
        return userEntity;
    }

    @StreamListener(StreamClient.STREAM_INPUT_2)
    public void process2(UserEntity userEntity) {
        log.info("STREAM_INPUT_2  input receiver: {}", userEntity.toString());
    }

yml

spring:
  rabbitmq:
    port: 5672
    addresses: 192.168.177.129
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        stream-input:
          destination: stream-output
          group: my-stream
          content-type: application/json
        stream-input-2:
          destination: stream-output-2
          group: my-stream
          content-type: application/json

项目地址

reference

Spring Cloud Stream如何消费自己生产的消息？

最后修改：2020 年 02 月 13 日

如果觉得我的文章对你有用，请随意赞赏

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款

评论 *

私密评论

名称 *

🎲

邮箱 *

地址

haha
属性配置列表属性名称描述选项MODE系统启动方式: 集群/单机...
haha
参数名描述可选项默认值MODE ...
金城武
老铁，请问你的伪静态文件在nginx里是怎么处理的？我的除了主...
龙猫
请问数据库呢
问问题的
init.sql貌似不生效

springboot+stream-rabbit学习笔记

hdy • 2019 年 07 月 02 日

引入pom.xml<pre><code> &lt;!-- https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-stream-rabbit --&gt;
 &lt;dependency&gt;
 &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
 &lt;artifactId&gt;spring-cloud-starter-stream-rabbit&lt;/artifactId&gt;
 &lt;version&gt;2.1.3.RELEASE&lt;/version&gt;
 &lt;/dependency&gt;</code></pre><ol><li>配置client<pre><code class="lang-java">/**
 * @author huangdeyao
 * @date 2019/7/2 12:50
 */
public interface StreamClient {
 /**
 * 同一个服务里面的信道名字不能一样，在不同的服务里可以相同名字的信道
 */
 String STREAM_INPUT = &quot;stream-input&quot;;
 String STREAM_OUTPUT = &quot;stream-output&quot;;

/**
  * 订阅消息
  *
  * @return
  */
 @Input(StreamClient.STREAM_INPUT)
 SubscribableChannel input();

/**
 * 消息发送
 *
 * @return
 */
 @Output(StreamClient.STREAM_OUTPUT)
 MessageChannel output();
}</code></pre></li><li>接收端<pre><code class="lang-java">/**
 * @author huangdeyao
 * @date 2019/7/2 13:04
 */
@Component
@EnableBinding(StreamClient.class)
@Log4j2
public class StreamReceiver {

@StreamListener(StreamClient.STREAM_INPUT)
 public void receive(Object message) {
 log.info(&quot;stream input receiver: {}&quot;, message);
 }
}
</code></pre></li><li>发送端<pre><code> @Autowired
 StreamClient streamClient;

@GetMapping(&quot;/send&quot;)
 public String messageWithMQ(String message) {
 streamClient.output().send(MessageBuilder.withPayload(message).build());
 return &quot;ok&quot;;
 }</code></pre></li><li>遇到的错误<pre><code>org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'stream_input' defined in com.dy.spring.client.StreamClient: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.dy.spring.client.StreamClient; factoryMethodName=input; initMethodName=null; destroyMethodName=null
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:68) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerOutputBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:55) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.lambda$registerBindingTargetBeanDefinitions$0(BindingBeanDefinitionRegistryUtils.java:92) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:410) ~[spring-core-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:389) ~[spring-core-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(BindingBeanDefinitionRegistryUtils.java:82) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.cloud.stream.config.BindingBeansRegistrar.registerBeanDefinitions(BindingBeansRegistrar.java:44) ~[spring-cloud-stream-2.1.3.RELEASE.jar:2.1.3.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.lambda$loadBeanDefinitionsFromRegistrars$1(ConfigurationClassBeanDefinitionReader.java:364) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[na:1.8.0_191]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:363) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:145) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:117) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:327) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:232) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:275) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:95) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:705) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:531) ~[spring-context-5.1.8.RELEASE.jar:5.1.8.RELEASE]
 at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:742) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:389) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.run(SpringApplication.java:311) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.run(SpringApplication.java:1213) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at org.springframework.boot.SpringApplication.run(SpringApplication.java:1202) [spring-boot-2.1.6.RELEASE.jar:2.1.6.RELEASE]
 at com.dy.spring.SpringDomeApplication.main(SpringDomeApplication.java:13) [classes/:na]

Disconnected from the target VM, address: '127.0.0.1:57769', transport: 'socket'

Process finished with exit code 1
</code></pre><blockquote>解决StreamClient 中不能通道名，使用不同的名字 修改</blockquote><pre><code>public interface StreamClient {
 /**
 * 同一个服务里面的信道名字不能一样，在不同的服务里可以相同名字的信道
 */
 String STREAM_INPUT = &quot;stream-input&quot;;
 String STREAM_OUTPUT = &quot;stream-output&quot;;
 
...... 略
}
yml配置,互相指向成一个通道stream-output
spring:
 cloud:
 stream:
 bindings:
 stream-input:
 destination: stream-output</code></pre></li><li>多实例接收多个消息解决，yml配置group<pre><code>spring:
 rabbitmq:
 port: 5672
 addresses: 192.168.177.129
 username: guest
 password: guest
 cloud:
 stream:
 bindings:
 stream-input:
 destination: stream-output
 group: my-stream</code></pre></li><li>接收后回调信息</li></ol><blockquote>StreamClient.class</blockquote><pre><code class="lang-java"> /**
 * 同一个服务里面的信道名字不能一样，在不同的服务里可以相同名字的信道
 */
 String STREAM_INPUT_2 = &quot;stream-input-2&quot;;
 String STREAM_OUTPUT_2 = &quot;stream-output-2&quot;;
 /**
 * 订阅消息
 *
 * @return
 */
 @Input(StreamClient.STREAM_INPUT_2)
 SubscribableChannel input2();

/**
 * 消息发送
 *
 * @return
 */
 @Output(StreamClient.STREAM_OUTPUT_2)
 MessageChannel output2();</code></pre><blockquote>StreamReceiver.class</blockquote><pre><code class="lang-java"> @StreamListener(StreamClient.STREAM_INPUT)
 @SendTo(StreamClient.STREAM_INPUT_2)
 public UserEntity process(UserEntity userEntity) {
 log.info(&quot;stream input receiver: {}&quot;, userEntity.toString());
 return userEntity;
 }

@StreamListener(StreamClient.STREAM_INPUT_2)
 public void process2(UserEntity userEntity) {
 log.info(&quot;STREAM_INPUT_2 input receiver: {}&quot;, userEntity.toString());
 }</code></pre><blockquote>yml</blockquote><pre><code>spring:
 rabbitmq:
 port: 5672
 addresses: 192.168.177.129
 username: guest
 password: guest
 cloud:
 stream:
 bindings:
 stream-input:
 destination: stream-output
 group: my-stream
 content-type: application/json
 stream-input-2:
 destination: stream-output-2
 group: my-stream
 content-type: application/json</code></pre><a class="no-external-link" href="https://gitee.com/dy.huang/spring-dome" target="_blank">项目地址</a><h2>reference</h2><a class="no-external-link" href="https://juejin.im/post/5bfe9df1e51d452e5e70f62a" target="_blank">Spring Cloud Stream如何消费自己生产的消息？</a>

springboot+stream-rabbit学习笔记

reference

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款

docker 搭建typecho

dockercompose搭建nacos集群

docker-compose + canal + mysql + redis

linux 使用某个角色运行程序

CentOS 7.0防火墙设置和关闭

MySQL5.7安装与配置

canal入手

docker-compose 安装redis

jdbc、jpa、spring data jpa、hibernate、mybatis之间的关系及区别

Linux Centos 7 - Nginx离线安装

springboot+stream-rabbit学习笔记

reference

发表评论 取消回复 使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款

springboot+stream-rabbit学习笔记

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款