代码编织梦想

阅读本文前可先参考

SpringCloud - Spring Cloud 之 Stream构建消息驱动微服务框架;RabbitMQ(十九)_MinggeQingchun的博客-CSDN博客_springcloud stream

一、Spring Cloud Stream

在微服务的开发过程中,会经常用到消息中间件,通过消息中间件在服务与服务之间传递消息,不管使用哪款消息中间件,如RabbitMQ、Kafka和RocketMQ,那么消息中间件和服务之间都有耦合性(如原来使用RabbitMQ,要替换为RocketMQ,那么微服务都需要修改,变动会比较大),因为这两款消息中间件有一些区别,如果我们Spring Cloud Stream来整合我们的消息中间件,就可以降低微服务和消息中间件的耦合性,做到轻松在不同消息中间件间切换

Spring Cloud Stream就是负责整合我们的消息中间件,降低微服务和消息中间件的耦合性,做到轻松在不同消息中间件间切换

官网地址:

Spring Cloud Stream

Spring Cloud Stream是一个框架,用于构建与共享消息系统连接的高度可伸缩的事件驱动微服务。 

Spring Cloud Stream解决了开发人员无感知的使用消息中间件的问题,Spring Cloud Stream对消息中间件的进一步封装,可以做到代码层面对消息中间件的无感知,甚至于动态的切换中间件(rabbitmq切换为rocketmq或者kafka),使得微服务开发的高度解耦。

注:

目前Spring Cloud Stream仅支持RabbitMQ、Kafka,Spring Cloud Alibaba新写了一个starter可以支持RocketMQ

Spring Cloud Stream架构 

Spring Cloud Stream 是一个构建消息驱动微服务的框架

应用程序通过Input(相当于消费者consumer)、Output(相当于生产者producer)来与Spring Cloud Stream中Binder交互,而Binder负责与消息中间件交互,因此,我们只需关注如何与Binder交互即可,而无需关注与具体消息中间件的交互

1、Binder

与外部消息中间件集成的组件,用来创建Binding,各消息中间件都有自己的 Binder 实现

Kafka 的实现 KafkaMessageChannelBinder

RabbitMQ 的实现 RabbitMessageChannelBinder

RocketMQ 的实现 RocketMQMessageChannelBinder

2、Binding

包括 Input Binding 和 Output Binding

Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁,实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可,屏蔽了开发者与底层消息中间件的接触

3、Input

应用程序通过input(相当于消费者consumer)与Spring Cloud Stream中Binder交互,而Binder负责与消息中间件交互,因此,我们只需关注如何与Binder交互即可,而无需关注与具体消息中间件的交互

4、Output

output(相当于生产者producer)与Spring Cloud Stream中Binder交互

组成

说明

Binder

Binder是应用与消息中间件之间的封装,目前实现了Kafka和RabbitMQ的Binder,通过Binder可以很方便的连接中间件,可以动态的改变消息类型(对应于Kafka的topic,RabbitMQ的exchange),这些都可以通过配置文件来实现

@Input

该注解标识输入通道,通过该输入通道接收消息进入应用程序

@Output

该注解标识输出通道,发布的消息将通过该通道离开应用程序

@StreamListener

监听队列,用于消费者的队列的消息接收

@EnableBinding

将信道channel和exchange、topic绑定在一起

二、Spring Cloud Stream应用(RocketMQ)

(一)Spring Cloud Stream应用(RocketMQ)

1、创建Springboot工程 springcloud-alibaba-4-stream-rocketmq

2、添加依赖 spring-cloud-starter-stream-rocketmq

<!--spring-cloud-starter-stream-rocketmq-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>
<groupId>com.bjpowernode</groupId>
    <artifactId>31-rocketmq-spring-cloud-stream</artifactId>
    <version>1.0.0</version>

    <name>31-rocketmq-spring-cloud-stream</name>
    <description>31-rocketmq-spring-cloud-stream project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.2.5.RELEASE</spring-boot.version>
        <spring-cloud-alibaba.version>2.2.1.RELEASE</spring-cloud-alibaba.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

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

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>${spring-cloud-alibaba.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.0.RELEASE</version>
                <configuration>
                    <mainClass>com.bjpowernode.Application</mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

3、application.properties配置文件

server.port=8081

spring.application.name=springcloud-alibaba-4-stream-rocketmq

# 日志级别
logging.level.com.alibaba.cloud.stream.binder.rocketmq=INFO


########## RocketMQ 通用配置
# 客户端接入点,必填,rocketmq的连接地址, binder高度抽象
spring.cloud.stream.rocketmq.binder.name-server=192.168.133.128:9876


########## 生产者Producer Config
# output 的配置如下: bindings 具体生产消息、消费消息的桥梁(消费者Consumer和生产者Producer的destination 目的地必须保持一致)
spring.cloud.stream.bindings.output.destination=test-topic
spring.cloud.stream.bindings.output.content-type=text/plain
spring.cloud.stream.bindings.output.group=test-group


########## 消费者Consumer Config
# input 的配置:(消费者Consumer和生产者Producer的destination 目的地必须保持一致)
spring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=text/plain
spring.cloud.stream.bindings.input.group=test-group

4、消息发送(生产者)

@Service
public class SenderService {

    //spring cloud stream里面发消息通过 Source 发送
    @Autowired
    private Source source;

    //原来springboot里面通过 RocketMQTemplate 发送

    /**
     * 发送消息的方法
     *
     * @param msg
     * @throws Exception
     */
    public void send(String msg) throws Exception {
        // source.output() == MessageChannel 消息通道
        boolean flag = source.output().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送:" + flag);
    }
}

5、消息接收(消费者)

@Service
public class ReceiveService {

    //spring cloud stream里面发消息通过 Sink 发送
    @Autowired
    private Sink sink;

    //原来springboot里面通过 RocketMQTemplate 发送

    /* 接收消息
    第一种:通过手动调用receive()方法接收消息,while循环监听消息
    * */
    public void receive() {
        while(true){
            // SubscribableChannel = sink.input() 消息订阅的信道
            sink.input().subscribe((Message<?> message) -> {
                System.out.println("Sink接收到的消息是:" + message.getPayload());
            });
        }
    }

    /* 接收消息
    第二种:通过@StreamListener监听消息,不需要调用receiveMessage(String message)方法 
    * */
    @StreamListener(value = Sink.INPUT)
    public void receiveMessage(String message) {
        System.out.println("StreamListener接收到的消息是:" + message);
    }
}

6、springboot启动程序类

import com.company.consumer.ReceiveService;
import com.company.producer.SenderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;

@EnableBinding(value = {Source.class, Sink.class}) //使其生效
@SpringBootApplication
//Spring boot的CommandLineRunner接口主要用于实现在应用初始化后,去执行一段代码块逻辑,这段初始化代码在整个应用生命周期内只会执行一次
public class Stream4RocketmqApplication implements CommandLineRunner {

    @Autowired
    private SenderService senderService;

    @Autowired
    private ReceiveService receiveService;


    public static void main(String[] args) {
        SpringApplication.run(Stream4RocketmqApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        senderService.send("Hello spring cloud stream rocketmq!");

        receiveService.receive();
    }
}

(二)Spring Cloud Stream自定义信道

消息传递主要使用的是系统提供的 Source (output)、Sink(input);因此我们自定义Source和Sink接口即可

server.port=8081

spring.application.name=springcloud-alibaba-4-stream-rocketmq

# 日志级别
logging.level.com.alibaba.cloud.stream.binder.rocketmq=INFO


########## RocketMQ 通用配置
# 客户端接入点,必填,rocketmq的连接地址, binder高度抽象
spring.cloud.stream.rocketmq.binder.name-server=192.168.133.128:9876


########## 生产者Producer Config
# output 的配置如下: bindings 具体生产消息、消费消息的桥梁(消费者Consumer和生产者Producer的destination 目的地必须保持一致)
spring.cloud.stream.bindings.output.destination=test-topic
spring.cloud.stream.bindings.output.content-type=text/plain
spring.cloud.stream.bindings.output.group=test-group

# output1 要对应到一个Source里面去
spring.cloud.stream.bindings.output1.destination=test-topic1
spring.cloud.stream.bindings.output1.content-type=text/plain
spring.cloud.stream.bindings.output1.group=test-group1

# output2 要对应到一个Source里面去
spring.cloud.stream.bindings.output2.destination=test-topic2
spring.cloud.stream.bindings.output2.content-type=text/plain
spring.cloud.stream.bindings.output2.group=test-group2

########## 消费者Consumer Config
# input 的配置:(消费者Consumer和生产者Producer的destination 目的地必须保持一致)
spring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=text/plain
spring.cloud.stream.bindings.input.group=test-group

spring.cloud.stream.bindings.input1.destination=test-topic1
spring.cloud.stream.bindings.input1.content-type=text/plain
spring.cloud.stream.bindings.input1.group=test-group1
spring.cloud.stream.rocketmq.bindings.input1.consumer.tags=myTag

spring.cloud.stream.bindings.input2.destination=test-topic2
spring.cloud.stream.bindings.input2.content-type=text/plain
spring.cloud.stream.bindings.input2.group=test-group2

消息发送(生产者)

public interface MySource {

    /**
     * Name of the output channel.
     */
    String OUTPUT1 = "output1";

    /**
     * Name of the output channel.
     */
    String OUTPUT2 = "output2";

    /**
     * @return output channel
     */
    @Output(MySource.OUTPUT1)
    MessageChannel output1();

    @Output(MySource.OUTPUT2)
    MessageChannel output2();
}

import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.util.MimeTypeUtils;

@Service
public class SenderService {

    //spring cloud stream里面发消息通过 Source 发送
    @Autowired
    private Source source;

    //原来springboot里面通过 RocketMQTemplate 发送

    @Autowired
    private MySource mySource;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送消息的方法
     *
     * @param msg
     * @throws Exception
     */
    public void send(String msg) throws Exception {
        // source.output() == MessageChannel 消息通道
        boolean flag = source.output().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送:" + flag);
    }

    /**
     * 发送消息的方法
     *
     * @param msg
     * @throws Exception
     */
    public void send(String msg) throws Exception {
        // source.output() == MessageChannel 消息通道
        boolean flag = source.output().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送:" + flag);
    }

    /**
     * 发送消息的方法
     *
     * @param msg
     * @throws Exception
     */
    public void send1(String msg) throws Exception {
        // source.output() == MessageChannel 消息通道
        boolean flag = mySource.output1().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送1:" + flag);
    }

    /**
     * 发送消息的方法,发到3个topic中
     *
     * @param msg
     * @throws Exception
     */
    public void multiSend(String msg) throws Exception {
        // source.output() == MessageChannel 消息通道
        boolean flag = source.output().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送:" + flag);

        // source.output() == MessageChannel 消息通道
        boolean flag1 = mySource.output1().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送1:" + flag1);

        // source.output() == MessageChannel 消息通道
        boolean flag2 = mySource.output2().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送2:" + flag2);
    }

    public <T> void sendObject(T msg, String tag) throws Exception {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();
        boolean flag2 = mySource.output1().send(message);

        System.out.println("对象消息发送2:" + flag2);
    }

    /**
     * 发送消息的方法
     *
     * @param msg
     * @throws Exception
     */
    public void sendTemplate(String msg) throws Exception {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();
        rocketMQTemplate.send("test-topic1", message);
        System.out.println("发送完毕......");
    }
}

消息接收(消费者)

public interface MySink {

    /**
     * Input channel name.
     */
    String INPUT1 = "input1";

    /**
     * Input channel name.
     */
    String INPUT2 = "input2";

    /**
     * @return input channel.
     */
    @Input(MySink.INPUT1)
    SubscribableChannel input1();

    /**
     * @return input channel.
     */
    @Input(MySink.INPUT2)
    SubscribableChannel input2();

}
@Service
public class ReceiveService {

    //spring cloud stream里面发消息通过 Sink 发送
    @Autowired
    private Sink sink;

    //原来springboot里面通过 RocketMQTemplate 发送

    @Autowired
    private MySink mySink;

    /* 接收消息
    第一种:通过手动调用receive()方法接收消息,while循环监听消息
    * */
    public void receive() {
        while(true){
            // SubscribableChannel = sink.input() 消息订阅的信道
            sink.input().subscribe((Message<?> message) -> {
                System.out.println("Sink接收到的消息是:" + message.getPayload());
            });
        }
    }

    /* 接收消息
    第二种:通过@StreamListener监听消息,不需要调用receiveMessage(String message)方法
    * */
    @StreamListener(value = Sink.INPUT)
    public void receiveMessage(String message) {
        System.out.println("StreamListener接收到的消息是:" + message);
    }

    public void receive1() {
        while (true){
            // SubscribableChannel = sink.input() 消息订阅的信道
            mySink.input1().subscribe((Message<?> message) -> {
                System.out.println("input 1---" + message.getPayload());
            });
        }
    }

    @StreamListener(value = MySink.INPUT1)
    public void receiveMessage1(String message) {
        System.out.println("接收到的消息是1:" + message);
    }

    @StreamListener(value = MySink.INPUT2)
    public void receiveMessage2(String message) {
        System.out.println("接收到的消息是2:" + message);
    }
}

(三)Spring Cloud Stream事务消息

#--------------------------事务消息-------------------------------------
#生产的配置
spring.cloud.stream.bindings.outputTX.destination=TransactionTopic
spring.cloud.stream.bindings.outputTX.content-type=application/json
spring.cloud.stream.rocketmq.bindings.outputTX.producer.group=myTxProducerGroup
#是否为事务消息,默认为false表示不是事务消息,true表示是事务消息
spring.cloud.stream.rocketmq.bindings.outputTX.producer.transactional=true

#消费的配置:
spring.cloud.stream.bindings.inputTX.destination=TransactionTopic
spring.cloud.stream.bindings.inputTX.content-type=text/plain
spring.cloud.stream.bindings.inputTX.group=transaction-group
spring.cloud.stream.rocketmq.bindings.inputTX.consumer.broadcasting=false

消息发送(生产者)

@Component
public class Sender {

    @Autowired
    private MySource mySource;

    public <T> void sendTransactionalMsg(T msg, int num) throws Exception {
        MessageBuilder builder = MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader("test", String.valueOf(num));
                //.setHeader(RocketMQHeaders.TAGS, "binder");
        Message message = builder.build();

        mySource.outputTX().send(message);
    }
}
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

@RocketMQTransactionListener(txProducerGroup = "myTxProducerGroup", corePoolSize = 5, maximumPoolSize = 10)
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

	/**
	 * 执行本地事务:也就是执行本地业务逻辑
	 *
	 * @param msg
	 * @param arg
	 * @return
	 */
	@Override
	public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
		Object num = msg.getHeaders().get("test");

		if ("1".equals(num)) {
			System.out.println("executer: " + new String((byte[]) msg.getPayload()) + " unknown");
			return RocketMQLocalTransactionState.UNKNOWN;
		}
		else if ("2".equals(num)) {
			System.out.println("executer: " + new String((byte[]) msg.getPayload()) + " rollback");
			return RocketMQLocalTransactionState.ROLLBACK;
		}
		System.out.println("executer: " + new String((byte[]) msg.getPayload()) + " commit");
		return RocketMQLocalTransactionState.COMMIT;
	}

	/**
	 * 回调检查
	 *
	 * @param msg
	 * @return
	 */
	@Override
	public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
		System.out.println("check: " + new String((byte[]) msg.getPayload()));
		return RocketMQLocalTransactionState.COMMIT;
	}
}

消息接收(消费者)

@StreamListener(value = MySink.INPUTTX)
    public void receiveTransactionMessage(String message) {
        System.out.println("接收到的 事务 消息是:" + message);
    }

三、Spring Cloud Stream RocketMQ配置选项

RocketMQ Binder Properties

spring.cloud.stream.rocketmq.binder.name-server

RocketMQ NameServer 地址(老版本使用 namesrv-addr 配置项)

Default: 127.0.0.1:9876.

spring.cloud.stream.rocketmq.binder.access-key

阿里云账号 AccessKey。

Default: null.

spring.cloud.stream.rocketmq.binder.secret-key

阿里云账号 SecretKey。

Default: null.

spring.cloud.stream.rocketmq.binder.enable-msg-trace

是否为 Producer 和 Consumer 开启消息轨迹功能

Default: true.

spring.cloud.stream.rocketmq.binder.customized-trace-topic

消息轨迹开启后存储的 topic 名称。

Default: RMQ_SYS_TRACE_TOPIC.

RocketMQ Consumer Properties

下面的这些配置是以 spring.cloud.stream.rocketmq.bindings.<channelName>.consumer. 为前缀的 RocketMQ Consumer 相关的配置。

enable

是否启用 Consumer

默认值: true.

tags

Consumer 基于 TAGS 订阅,多个 tag 以 || 分割

默认值: empty.

sql

Consumer 基于 SQL 订阅

默认值: empty.

broadcasting

Consumer 是否是广播消费模式。如果想让所有的订阅者都能接收到消息,可以使用广播模式

默认值: false.

orderly

Consumer 是否同步消费消息模式

默认值: false.

delayLevelWhenNextConsume

异步消费消息模式下消费失败重试策略:

-1,不重复,直接放入死信队列

0,broker 控制重试策略

>0,client 控制重试策略

默认值: 0.

suspendCurrentQueueTimeMillis

同步消费消息模式下消费失败后再次消费的时间间隔

默认值: 1000.

RocketMQ Provider Properties

下面的这些配置是以 spring.cloud.stream.rocketmq.bindings.<channelName>.producer. 为前缀的 RocketMQ Producer 相关的配置

enable

是否启用 Producer

默认值: true.

group

Producer group name

默认值: empty.

maxMessageSize

消息发送的最大字节数

默认值: 8249344.

transactional

是否发送事务消息

默认值: false.

sync

是否使用同步得方式发送消息

默认值: false.

vipChannelEnabled

是否在 Vip Channel 上发送消息

默认值: true.

sendMessageTimeout

发送消息的超时时间(毫秒)

默认值: 3000.

compressMessageBodyThreshold

消息体压缩阀值(当消息体超过 4k 的时候会被压缩)

默认值: 4096.

retryTimesWhenSendFailed

在同步发送消息的模式下,消息发送失败的重试次数

默认值: 2.

retryTimesWhenSendAsyncFailed

在异步发送消息的模式下,消息发送失败的重试次数

默认值: 2.

retryNextServer

消息发送失败的情况下是否重试其它的 broker

默认值: false

由此开发使用RocketMQ有两种选择

1、SpringBoot + RocketMQ整合实现消息传送;

2、使用Spring Cloud Stream对消息中间件的包装,来实现消息传送

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/MinggeQingchun/article/details/126277831

使用 spring cloud stream 构建消息驱动微服务_xiaomin_____的博客-爱代码爱编程

微服务的目的: 松耦合 事件驱动的优势:高度解耦 Spring Cloud Stream 的几个概念 Spring Cloud Stream is a framework for building message-driven microservice applications.

「 从0到1学习微服务springcloud 」08 构建消息驱动微服务的框架 spring cloud stream_ken do everthing的博客-爱代码爱编程

系列文章(更新ing): 「 从0到1学习微服务SpringCloud 」01 一起来学呀! 「 从0到1学习微服务SpringCloud 」02 Eureka服务注册与发现 「 从0到1学习微服务SpringCloud

springCloud-Alibaba——stream+rocketmq消息微服务集成-爱代码爱编程

目录 (1)引入依赖: (2)修改启动类: (3)修改配置文件: A、生产者配置文件: B、消费者配置文件: (4)编写代码: A、发送消息代码: B、消费消息代码: (5)启动测试: (1)引入依赖: 在原有springCloud、springCloud-Alibaba管理依赖基础上,增加stream-rocketmq的依赖包:

(未完待续)浅谈微服务以及 常用中间件( zookeeper redis rabbitmq)-爱代码爱编程

   传统的单体框架,已经不满足目前公司战略规划要求,近几年“微服务“ 这个字眼,出现的越来越频繁,虽然有过一年多微服务项目经验,也很难把微服务解释清楚,到底何为微服务?    Martin Fowler曾在其blog上发表了”Microservices“的文章,正式提出了微服务架构,对于微服务的解释,他的定义为:微服务架构是一种架构模式,他 提倡将单一

流式消息_spring-cloud-starter-stream-rocketmq发送消息_v1.0.1-爱代码爱编程

文件名称版本号作者qq组件版本流式消息_spring-cloud-starter-stream-rocketmq发送消息v1.0.1学生宫布8416837SpringBoot 2.1.8.RELEASE spring-cloud-starter-stream-rocketmq 2.1.1.RELEASE 文章目录 简介说明模式发布代码订阅 同理 不过

Spring Cloud Stream应用RocketMQ-爱代码爱编程

What is Spring Cloud Stream?   微服务中会经常使用消息中间件,通过消息中间件在服务与服务之间传递消息,例如RabbitMQ、Kafka和RocketMQ,无论使用哪一种消息中间件和服务之间都有一点耦合性,这个耦合性指的是原来使用RabbitMQ,现在要替换为RocketMQ,我们的微服务改动比较大,因为两款消息中间件有一些区

消息驱动 微服务器,消息驱动的微服务-Spring Cloud Stream整合RocketMQ-爱代码爱编程

系列文章导航: Spring Cloud Alibaba微服务解决方案 常用MQ产品的选择 目前主流的MQ产品有kafka、RabbitMQ、ActiveMQ、RocketMQ等。在MQ选型时可以参照这篇文章选择合适的MQ产品。 RocketMQ及控制台搭建 RocketMQ的搭建可以参考这篇文章。 RocketMQ控制台的搭建可以参考这篇文

【Spring Cloud】12.SpringCloud Stream消息驱动-爱代码爱编程

目录 1.简介1.1 概括1.2 思想理解1.2.1 基础MQ1.2.2 使用Spring Cloud Stream2.建立项目2.1 cloud-stream-rabbitmq-provider88012.1.1 Pom.xml2.1.2 bootstrap.yml2.1.3 Service层2.1.3.1 IMessageProvider.ja

spring cloud alibaba开发笔记十一(基于SpringCloud Stream构建消息驱动微服务)-爱代码爱编程

SpringBoot集成Kafka构建消息驱动微服务 下载、安装Kafka ◆下载Kafka : https://kafka.apache.org/quickstart ◆解压、启动ZK和Kafka Server即可(使用默认配置) SpringCloud Stream消息驱动组件概览 ◆负责与中间件交互的抽象绑定器: Binde

SpringCloudStream集成RocketMQ-爱代码爱编程

1.1 添加依赖 SpringCloudStream集成RocketMQ,Alibaba按照SpringCloudStream的规范实现了将信息推送给发送和接收通道的绑定器 在消息发送方和订阅方都添加依赖 <!-- SpringCloudStream 集成RocketMQ --> <dependency> <

springcloud alibaba stream 集成rocketmq_罗密欧没有朱丽叶的博客-爱代码爱编程

说明 本示例只是简单仅是入门级的示例,详细使用后续会持续更新 测试版本: springcloud :Hoxton.SR8springcloud alibaba: 2.2.3.RELEASEspringboot : 2.3.2.RELEASErockectMQ: V4.4.0 Springcloud Stream交互示意图 搭建 Dem

spring-爱代码爱编程

Spring Cloud Stream 在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务,其目的是为了简化消息在 Spring Cloud 应用程序中的开发。 Spring Cloud Stream的核心构件是: Destination Binders: 负责提供与外部消息系统集成的组件。 Destination Bindi

springcloudstream 集成rocketmq_spring-爱代码爱编程

目录 一、RocketMQ可视化界面安装 二、springcloudstream整合rocketmq 1、引入依赖 2、开启输入何输出 3、关键的一步,配置文件 4、主类 5、控制器 6、消息消费者 7、消息生产者 RocketMQ安装步骤略。 开启rocketmq: windows start mqnamesrv.cmd

springcloud -爱代码爱编程

阅读本文前可先参考 ​​​​​​SpringCloud - Spring Cloud根/父项目,开发准备(二)_MinggeQingchun的博客-CSDN博客 在微服务的开发过程中,经常用到消息中间件,通过消息中间件在服务与服务之间传递消息,不管使用哪款消息中间件,如RabbitMQ还是Kafka,那么消息中间件和服务之间都有耦合性,如原来使用的