SpringBoot整合SpringCloudStream3.1+版本Kafka

下一节直通车

SpringBoot整合SpringCloudStream3.1+版本的Kafka死信队列

为什么用SpringCloudStream3.1

Springcloud架构提供，基于spring生态
能够快速切换市面上常见的MQ产品
3.1后使用配置文件的形式定义channel，不再需要3.1前的硬编码

Jar

<!--SpringCloudStream Kafka-->
<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-starter-stream-kafka</artifactId>
	<version>3.2.4</version>
</dependency>
<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-stream</artifactId>
	<version>3.2.4</version>
</dependency>
<!--Kafka相关-->
<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
	<version>2.8.6</version>
</dependency>
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-web</artifactId>
	<version>2.6.8</version>
</dependency>
<dependency>
	<groupId>org.projectlombok</groupId>
	<artifactId>lombok</artifactId>
	<version>1.18.24</version>
</dependency>

配置文件yml

kafka基本配置（application-mq.yml）

server:
  port: 7105
spring:
  application:
	name: betrice-message-queue
  config:
	import:
	- classpath:application-bindings.yml
  cloud:
	stream:
	  kafka:
		binder:
		  brokers: localhost:9092
		  configuration:
			key-serializer: org.apache.kafka.common.serialization.StringSerializer
			value-serializer: org.apache.kafka.common.serialization.StringSerializer
		  consumer-properties:
			enable.auto.commit: false
	  binders:
		betrice-kafka:
		  type: kafka
		  environment:
			spring.kafka:
			  bootstrap-servers: ${spring.cloud.stream.kafka.binder.brokers}

classpath:application-bindings.yml 引入通道绑定配置文件，即消息生产、消费者的关系。

通道绑定配置（application-bindings.yml）

参数含义：
spring.cloud.stream.function.definition：定义channel名字，每个channel又可以作为生产者（in）与消费者（out）
spring.cloud.stream.bindings: 是一个map形式（具体看源码的Properties定义）
1. destination（生产/消费者通向的topic）；
2. group：消费者组名；
3. binder：绑定当前使用的MQ类型（见betrice-kafka）；
4. content-type：消息序列/反序列化的类型（见源码的支持的类型）

spring:
  cloud:
	stream:
	  betrice-default-binder: betrice-kafka
	  function:
		# 声明两个channel，transfer接收生产者的消息，处理完后给sink
		definition: transfer;sink;gather;gatherEcho
	  bindings:
		# 添加生产者binding，输出到destination对应的topic
		Evad05:
		  destination: Evad07
		  binder: ${spring.cloud.stream.betrice-default-binder}
		transfer-in-0:
		  destination: Evad07
		  binder: ${spring.cloud.stream.betrice-default-binder}
		transfer-out-0:
		  destination: Evad08
		  binder: ${spring.cloud.stream.betrice-default-binder}
		  content-type: text/plain
		sink-in-0:
		  destination: Evad08
		  binder: ${spring.cloud.stream.betrice-default-binder}
		  content-type: text/plain

Controller

/**
 * @author Evad.Wu
 * @Description 消息队列 控制类
 * @date 2023-02-10
 */
@Slf4j
@RestController
@RequestMapping(value = "betriceMqController")
public class BetriceMqController {
	@Resource(name = "betriceKafkaProducer")
	private Producer<String, String> producer;
	@Resource(name = "streamBridgeUtils")
	private StreamBridge streamBridge;

	@PostMapping("send")
	public void send(String topic, String key, String message) {
		try {
			producer.send(new ProducerRecord<>(topic, key, message));
			log.info("发送消息：" + message);
		} catch (Exception e) {
			log.error("异常消息：" + e);
		}
	}

	@PostMapping("streamSend")
	public void streamSend(String topic, String message) {
		try {
			streamBridge.send(topic, message);
			log.info("发送消息：" + message);
		} catch (Exception e) {
			log.error("异常消息：" + e);
		}
	}
}

Channel（生产、消费者通道）

/**
 * @author Evad.Wu
 * @Description mq消息通道 配置类
 * @date 2023-02-11
 */
@Configuration
public class BetriceMqSubChannel {
	/**
	 * Function方式声明binding，相当于同时声明了一个Producer的Bindng和一个Consumer的Binding。
	 * transfer-in-0 表示消费者  transfer-out-0 表示生产者
	 * 其作用就在于，echo-in-0收到的消息，立即就会通过echo-out-0发送出去。
	 */
	@Bean
	public Function<String, String> transfer() {
		return message -> {
			System.out.println("transfer: " + message);
			throw new RuntimeException("死信队列测试！");
//            return "transfer:" + message;
		};
	}

	/**
	 * Consumer声明一个消息消费者，sink1就对应sink-in-0
	 */
	@Bean
	public Consumer<String> sink() {
		return message -> {
			System.out.println("******************");
			System.out.println("At Sink1");
			System.out.println("******************");
			System.out.println("Received message " + message);
		};
	}
}