SpringCloud 集成 RocketMQ 及配置解析

码农世界 2024-06-12 后端 88 次浏览 0个评论

文章目录

前言
一、SpringCloud 集成 RocketMQ

1. pom 依赖
2. yml 配置
3. 操作实体
4. 生产消息

4.1. 自动发送消息
4.2. 手动发送消息
5. 消费消息
二、配置解析

1. spring.cloud.function.definition 方法定义
2. spring.cloud.stream.binders 公共服务配置
3. spring.cloud.stream.bindings 公共消息通道配置
2. spring.cloud.stream.rocketmq.binder RocketMQ 服务配置

3. spring.cloud.stream.rocketmq.bindings RocketMQ 消息通道配置

前言

定义
Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于 Spring Boot 来创建独立的、可用于生产的 Spring 应用程序。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，并引入了发布-订阅、消费组、分区这三个核心概念。简单的说，Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration，实现了一套轻量级的消息驱动的微服务框架。
抽象模型
我们都知道市面上有很多消息中间件，Sping Cloud Stream 为了可以集成各种各样的中间件，它抽象出了 Binder 的概念,每个消息中间件都需要有对应自己的 Binder。这样它就可以根据不同的 Binder 集成不同的中间件。下图的input和output是channel，Binder则是消息中间件和通道之间的桥梁。
绑定器
通过使用 Spring Cloud Stream，可以有效简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。但是目前 Spring Cloud Stream 只支持 RabbitMQ 和 Kafka 的自动化配置。
Spring Cloud Stream 提供了 Binder (负责与消息中间件进行交互)，我们则通过 inputs 或者 outputs 这样的消息通道与 Binder 进行交互。

Binder 绑定器是 Spring cloud Stream 中一个非常重要的概念，实现了应用程序和消息中间件之间的隔离，同时我们也可以通过应用程序实现，消息中间件之间的通信。在我们的项目的可以继承多种绑定器，我们可以根据不同特性的消息使用不同的消息中间件。Spring Cloud Stream 为我们实现了 RabbitMQ 和Kafka 的绑定器。如果你想使用其他的消息中间件需要自己去实现绑定器接口。

一、SpringCloud 集成 RocketMQ

1. pom 依赖



    com.alibaba.cloud
    spring-cloud-starter-stream-rocketmq

2. yml 配置

spring:
  cloud:
    stream:
	  function:
	    definition: producer1;consumer1 # 方法定义（用于定义发送者或消费者方法，多个分号隔开）
      # 配置消息通道通用属性（适用于所有消息中间件）
      bindings:
        producer1-out-0:
          destination: producer_topic # topic消息主题
          content-type: application/json # 内容格式
        # 配置channel消息通道
        consumer1-in-0:
          destination: consumer_topic # topic消息主题
          content-type: application/json # 内容格式
          group: consumer-group # 消费者组
      rocketmq:
        binder:
          name-server: 127.0.0.1:9876 # rocketmq服务地址
          vipChannelEnabled: true # 是否开启vip通道（兼容老版本使用。多监听一个端口用于接受处理消息，防止端口占用。）
        # 配置消息通道独特属性（仅适用于rocketmq）
        bindings:
          # 配置channel消息通道（生产者：[functionName]-out-[index]，消费者：[functionName]-in-[index]）
          producer1-out-0:
            producer:
              group: consumer-group
              sync: true # 是否开启同步发送
          consumer1-in-0: 
            consumer:
              subscription: myTag  # 消费tag
              delayLevelWhenNextConsume: -1
              suspendCurrentQueueTimeMillis: 99999999
              broadcasting: false # 是否使用广播消费，默认为false使用集群消费

3. 操作实体

package com.demo.model;
import lombok.AllArgsConstructor;
import lombok.Data;
/**
 * 消息model
 */
@Data
@AllArgsConstructor
public class MsgModel {
    /**
     * 消息id
     */
    private String msgId;
    /**
     * 消息内容
     */
    private String message;
}

4. 生产消息

4.1. 自动发送消息

通过 MessageBuilder 自动发送消息。

package com.demo;
import com.demo.model.MsgModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import java.util.function.Supplier;
/**
 * 消息生产者类
 */
@Configuration
@Slf4j
public class MyProducer {
    
	/**
	 * 消息生产者1
	 */
    @Bean
    public Supplier> producer1() {
        return () -> {
            MsgModel msgModel = new MsgModel(System.currentTimeMillis(), "测试消息");
            log.info("producer1发送消息：" + msgModel);
            return MessageBuilder.withPayload(entity).build();
        };
    }
}

这种方式定义 suppelier 会默认1000ms 发送一次记录。可以修改 spring.cloud.stream.poller.fixedDelay 设置延迟毫秒值。

4.2. 手动发送消息

通过 StreamBridge 手动发送消息。

package com.demo.controller;
import com.demo.model.MsgModel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
/**
 * 消息controller
 */
@RestController
@RequiredArgsConstructor
@RequestMapping("/msg")
@Slf4j
public class MsgController {
    private final StreamBridge streamBridge;
	/**
	 * 发送消息
	 */
    @GetMapping("/send")
    public void sendMsg() {
        MsgModel msgModel = new MsgModel(System.currentTimeMillis(), "测试消息");
        log.info("producer1发送消息：" + msgModel);
        streamBridge.send("producer1-out-0", 
        	MessageBuilder.withPayload(entity).setHeader("MyHearder", "这是一个请求头").build());
    }
}

5. 消费消息

package com.demo;
import com.demo.model.MsgModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import java.util.function.Consumer;
/**
 * 消息消费者类
 */
@Configuration
@Slf4j
public class ReceiveMQ {
	/**
	 * 消息消费者1
	 */
    @Bean
    public Consumer> consumer1(){
        return (message)->{
            MessageHeaders headers = message.getHeaders();
            MsgModel msgModel = message.getPayload();
            log.info("consumer1接收消息，消息头：" + headers.get("MyHeader"));
            log.info("consumer1接收消息，消息内容：" + msgModel);
        };
    }
}

二、配置解析

1. spring.cloud.function.definition 方法定义

进行生产者或消费者方法定义，在 rocketmq 初始时会加载这些方法以创建生产者或消费者列表。

不管是创建 Consumer 还是 Supplier 或者是 Function Stream 都会将其方法名称进行一个 topic 拆封和绑定。假设创建了一个 Consumer< String > myTopic 的方法，Stream 会将其拆分成 In 和 out 两个通道：

输入通道（消费者）： [functionName]-in-[index]
consumer1-in-0

输出通道（生产者）： [functionName]-out-[index]

producer1-out-0

注意：这里的 functionName 需要和生产者或消费者方法名称以及 spring.cloud.function.definition 下的名称保持一致。

spring:
  cloud:
    stream:
	  function:
	    definition: producer1;consumer1 # 方法定义（用于定义发送者或消费者方法，多个分号隔开）

2. spring.cloud.stream.binders 公共服务配置

配置消息中间件公共的连接信息。

配置消息中间件公共的主机名、端口、用户名和密码等信息。可以配置多个binder，每个binder对应一个消息中间件。

当需要定义多个消息中间件时，使用 stream.binders 定义。

当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.binder 定义。

spring:
  cloud:
    stream:
      # 公共服务配置
      # 当需要定义多个消息中间件时，使用 stream.binders 定义。
      # 当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.binder 定义。
      binders:
        rocket1:
          type: rocket # 消息中间件类型
          environment:
            spring:
              rocketmq:
                host: 127.0.0.1 # IP地址
                port: 9876		# 端口
                username: guest	# 账号
                password: guest # 密码

3. spring.cloud.stream.bindings 公共消息通道配置

配置消息通道公共的配置信息。

配置消息通道公共的名称、目标（对应消息中间件中的队列或主题名）、分区、消息内容类型等信息。

当需要定义多个消息中间件时，使用 stream.bindings 定义。

当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.bindings 定义。

spring:
  cloud:
    stream:
      # 公共消息通道配置
      # 当需要定义多个消息中间件时，使用 stream.bindings 定义。
      # 当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.bindings 定义。
      # 定义生产者通道：[functionName]-out-[index]、定义消费者通道：[functionName]-in-[index]。
      bindings:
        # 生产者通道
        producer1-out-0:
          destination: producer1_topic # 消息队列（topic主题）
          content-type: application/json # 消息内容格式
          binder: rocket1 # 绑定的消息中间件
        # 消费者通道
        consumer1-in-0:
          destination: consumer1_topic # 消息队列（topic主题）
          content-type: application/json # 消息内容格式
          group: consumer1-group # 消费组（在一个组内，一条消息只能被消费一次）
          binder: rocket1 # 绑定的消息中间件

2. spring.cloud.stream.rocketmq.binder RocketMQ 服务配置

配置 RocketMQ 消息中间件的连接信息。

配置 RocketMQ 消息中间件的主机名、端口、用户名和密码等信息。

spring:
  cloud:
    stream:
      rocketmq:
      	# 服务配置
        binder:
          name-server: 127.0.0.1:9876 # 服务地址
          vipChannelEnabled: true # 是否开启vip通道（兼容老版本使用。多监听一个端口用于接受处理消息，防止端口占用）

3. spring.cloud.stream.rocketmq.bindings RocketMQ 消息通道配置

配置 RocketMQ 消息通道的配置信息。

配置 RocketMQ 通道的名称、目标（对应消息中间件中的队列或主题名）、分区、消息内容类型等信息。

spring:
  cloud:
    stream:
      rocketmq:
        bindings:
          # 生产者通道（可定义通道的生产者producer）
          producer1-out-0:
            producer:
              group: producer1-group # 生产组（多个生产者实例，当一个实例挂掉时，使用另一个实例执行消息的提交或回滚）
              sync: true # 是否同步发送（发送消息后同步等待，直至发送结果返回）
          # 消费者通道（可定义通道的消费者consumer）
          consumer1-in-0: 
            consumer:
              group: # 消费组（多个消费者实例，用于负载均衡和容错。每个实例订阅的主题必须相同）
              subscription: myTag  # 消费tag（多个tag用||隔开）
              broadcasting: false # 是否使用广播消费，默认为false使用集群消费
              maxReconsumeTimes: 16 # 最大重试次数（-1 – 16，默认16）
              delayLevelWhenNextConsume: -1 # 重试消费的延迟等级（-1不重试；默认0，默认等级3）
              suspendCurrentQueueTimeMillis: 99999999 # 重试时间间隔ms（默认1000）

转载请注明来自码农世界，本文标题：《SpringCloud 集成 RocketMQ 及配置解析》