一. 首先安装rabbitmq-management

这里用的是rabbitmq的docker镜像，我们可以在Docker Hub中搜索rabbitmq, 找到最新的版本安装

sudo docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3.7.8-management

安装之后使用

docker ps -a

检查下, rabbitmq的镜像是否启动, 正常启动状态如下:

通过http://192.168.12.12:15672, 访问到rabbitmq的管理端,
默认账户/密码是: guest/guest

二. Spring Cloud Stream与RabbitMQ集成

引入依赖

<!-- Spring Cloud Stream RabbitMQ -->
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

创建消息接受类

@Component
@Slf4j
@EnableBinding(Processor.class)
public class MyMQReciver {@StreamListener(Processor.INPUT)public void process(String message){log.info("hahahah : "+message);System.out.println("hahahah : "+message);}
}

代码解读两处:
1. @StreamListener(Processor.INPUT)
这里其实是要声明一个订阅的键值, Processor类是一个org.springframework.cloud.stream.messaging jar包中内置的接口，查看其源码可以看到它继承了Source和Sink两个类

package org.springframework.cloud.stream.messaging;public interface Processor extends Source, Sink {
}

package org.springframework.cloud.stream.messaging;import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;public interface Source {String OUTPUT = "output";@Output("output")MessageChannel output();
}

package org.springframework.cloud.stream.messaging;import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;public interface Sink {String INPUT = "input";@Input("input")SubscribableChannel input();
}

结构很简单, 我们也可以仿照去实现自己的Processor
2. @EnableBinding(Processor.class)
这里绑定的就是Processor(或者我们自己实现的Processor)

创建发送消息测试类

@RestController
public class SendController {@Autowiredprivate Processor pipe;@GetMapping("/send")public void send(@RequestParam String message){pipe.output().send(MessageBuilder.withPayload(message).build());}
}

在application.yml中增加配置

spring:cloud:stream:
      bindings:
        input:
          destination: jeecgCloudExchangebinder: local_rabbitgroup: logMessageConsumersoutput:
          destination: jeecgCloudExchangebinder: local_rabbitbinders:
        local_rabbit:
          type: rabbitenvironment:
            spring:
              rabbitmq:
                host: localhostport: 5672username: guestpassword: guestvirtual-host: /

启动SpringCloud项目

首先浏览器进入rabbitmq管理端查看, 发现我们在application.yml中创建的output destination被自动创建出来了

input destination也被自动创建出来了,并且自动添加了绑定

确认rabbitmq这边没有问题后, 我们通过访问消息接口测试， http://localhost:8001/send?message=hello
发现MyMQReciver已经成功接受到了消息

三. 理论知识点

Spring Cloud Stream核心架构1

两个比较重要的地方：inputs(输入)消息接收端、outputs(输出)消息发送端

一个 Spring Cloud Stream 应用以消息中间件为核心，应用通过Spring Cloud Stream注入的输入/输出通道 channels 与外部进行通信。channels 通过特定的Binder实现与外部消息中间件进行通信。

Spring Cloud Stream核心架构2

黄色：表示RabbitMQ

绿色：插件，消息的输入输出都套了一层插件，插件可以用于各种各样不同的消息，也可以用于消息中间件的替换。

核心概念：

Barista接口：Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称，通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。

通道接口如何定义：

@Output：输出注解，用于定义发送消息接口

@Input：输入注解，用于定义消息的消费者接口

@StreamListener：用于定义监听方法的注解

使用Spring Cloud Stream 非常简单，只需要使用好这3个注解即可，在实现高性能消息的生产和消费的场景非常合适，但是使用SpringCloudStream框架有一个非常大的问题,就是不能实现可靠性的投递，也就是没法保证消息的100%可靠性，会存在少量消息丢失的问题。目前SpringCloudStream整合了RabbitMQ与Kafka，我们都知道Kafka是无法进行消息可靠性投递的，这个原因是因为SpringCloudStream框架为了和Kafka兼顾所以在实际工作中使用它的目的就是针对高性能的消息通信的！这点就是在当前版本SpringCloudStream的定位。

因此在实际的工作中，可以采用SpringCloudStream，如果需要保证可靠性投递，也可以单独采用RabbitMQ，也是可以的。

介绍Spring Cloud Stream与RabbitMQ集成相关推荐

使用Spring Cloud Stream与RabbitMQ集成
在我以前的文章中,我写了两个系统之间非常简单的集成场景-一个生成一个工作单元,另一个处理该工作单元,以及Spring Integration如何使这种集成非常容易. 在这里,我将演示如何使用Sprin ...
Spring Cloud Stream与RabbitMQ 死信队列
RabbitMQ的TTL全称为Time-To-Live,表示的是消息的有效期.消息如果在队列中一直没有被消费并且存在时间超过了TTL,消息就会变成"死信" (Dead Messag ...
Spring Cloud Stream多RabbitMQ实例配置时报错no default binder has been set
当前Spring Cloud Rabbit的版本为2.1.2 <dependency><groupId>org.springframework.cloud</groupI ...
【本人秃顶程序员】使用Spring Cloud Stream和RabbitMQ实现事件驱动的微服务
←←←←←←←←←←←← 快!点关注让我们展示如何使用Spring Cloud Stream来设计事件驱动的微服务.首先,Spring Cloud Stream首先有什么好处?因为Spring AM ...
Spring Cloud Stream与RabbitMQ 生产者和消费者位于同一个应用服务
第一种模型:交换机类型为topic,路由key为"#",这是简单的使用模型当前Spring Cloud Rabbit的版本为2.1.2 <dependency>< ...
Spring Cloud Stream与RabbitMQ 消费者消息分组
Spring Cloud Stream 其实是发布订阅模型,如果一个topic有多个订阅实例 ,消息就会被这些消息消费者接收到,这样就会带来一个问题,那就是消息的重复消费,这种问题在很多业务场景下是不 ...
Spring Cloud Stream与RabbitMQ整合
Spring Cloud与RabbitMQ整合需要引入下述依赖 <dependency><groupId>org.springframework.cloud</group ...
Spring Cloud Stream与RabbitMQ整合时Producer与Consumer的相关配置
生产者属性下面的属性都必须添加前缀: spring.cloud.stream.<rabbitName>.bindings.<channelName>.producer. 如果 ...
Spring Cloud Stream整合RabbitMQ
生产者: 消费者:

介绍Spring Cloud Stream与RabbitMQ集成

一. 首先安装rabbitmq-management

二. Spring Cloud Stream与RabbitMQ集成

三. 理论知识点

介绍Spring Cloud Stream与RabbitMQ集成相关推荐

最新文章

热门文章