springCloudStream消息流demo

xjs-study/springcloud-project/pom.xml

@@ -12,6 +12,7 @@
     <name>微服务组件</name>
     <modules>
         <module>eureka</module>
+        <module>stream</module>
     </modules>
 
     <artifactId>springcloud-project</artifactId>

xjs-study/springcloud-project/stream/pom.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>springcloud-project</artifactId>
+        <groupId>com.xjs</groupId>
+        <version>1.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <packaging>pom</packaging>
+    <name>Stream消息流</name>
+    <modules>
+        <module>stream-producer-8600</module>
+        <module>stream-consumer-8601</module>
+    </modules>
+
+    <artifactId>stream</artifactId>
+
+    <properties>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+    </properties>
+
+</project>

xjs-study/springcloud-project/stream/stream-consumer-8601/pom.xml

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>stream</artifactId>
+        <groupId>com.xjs</groupId>
+        <version>1.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <name>Stream消费者8600</name>
+
+    <artifactId>stream-consumer-8601</artifactId>
+
+    <properties>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

xjs-study/springcloud-project/stream/stream-consumer-8601/src/main/java/com/xjs/StreamConsumerApp8601.java

@@ -0,0 +1,18 @@
+package com.xjs;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+
+/**
+ * @author xiejs
+ * @since 2022-05-26
+ */
+@SpringBootApplication
+@EnableDiscoveryClient
+public class StreamConsumerApp8601 {
+
+    public static void main(String[] args) {
+        SpringApplication.run(StreamConsumerApp8601.class, args);
+    }
+}

xjs-study/springcloud-project/stream/stream-consumer-8601/src/main/java/com/xjs/service/StreamConsumerService.java

@@ -0,0 +1,20 @@
+package com.xjs.service;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Service;
+
+import java.util.function.Consumer;
+
+/**
+ * @author xiejs
+ * @since 2022-05-26
+ */
+
+@Service
+public class StreamConsumerService {
+
+    @Bean
+    public Consumer<String> myChannel() {
+        return message -> System.out.println("消息："+message);
+    }
+}

xjs-study/springcloud-project/stream/stream-consumer-8601/src/main/resources/application.yml

@@ -0,0 +1,35 @@
+server:
+  port: 8601
+
+spring:
+  application:
+    name: stream-consumer
+
+  cloud:
+    stream:
+      binders:    #绑定mq服务信息
+        rabbitBinder:   #名称定义，用于后面的关联
+          type: rabbit
+          environment:    #mq环境配置
+            spring:
+              rabbitmq:
+                host: 192.168.23.128
+                port: 5672
+                username: guest
+                password: guest
+
+      bindings: # 关联整合通道和binder对象
+        myChannel-in-0:  # output是我们定义的通道名称，此处不能乱改
+          destination: exchange # 要使用的Exchange名称（消息队列主题名称）
+          content-type: application/json # application/json # 消息类型设置，比如json
+          binder: rabbitBinder # 关联MQ服务
+          group: default
+eureka:
+  client:
+    serviceUrl: # eureka server的路径
+      defaultZone: http://localhost:8761/eureka,http://localhost:8762/eureka,http://localhost:8763/eureka
+    instance:
+      prefer-ip-address: true #使用ip注册
+      instance-id: ${spring.cloud.client.ip-address}:${spring.application.name}:${server.port}   #实例名称
+
+

xjs-study/springcloud-project/stream/stream-producer-8600/pom.xml

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>stream</artifactId>
+        <groupId>com.xjs</groupId>
+        <version>1.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <name>Stream生产者8600</name>
+
+    <artifactId>stream-producer-8600</artifactId>
+
+    <properties>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

xjs-study/springcloud-project/stream/stream-producer-8600/src/main/java/com/xjs/StreamProducerApp8600.java

@@ -0,0 +1,18 @@
+package com.xjs;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+
+/**
+ * @author xiejs
+ * @since 2022-05-26
+ */
+@SpringBootApplication
+@EnableDiscoveryClient
+public class StreamProducerApp8600 {
+
+    public static void main(String[] args) {
+        SpringApplication.run(StreamProducerApp8600.class, args);
+    }
+}

xjs-study/springcloud-project/stream/stream-producer-8600/src/main/java/com/xjs/controller/MessageProducerController.java

@@ -0,0 +1,26 @@
+package com.xjs.controller;
+
+import com.xjs.service.IMessageProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xiejs
+ * @since 2022-05-26
+ */
+@RestController
+@RequestMapping("message")
+public class MessageProducerController {
+    @Autowired
+    private IMessageProducer iMessageProducer;
+
+
+    @GetMapping
+    public String message() {
+        iMessageProducer.sendMessage();
+
+        return "xaxa";
+    }
+}

xjs-study/springcloud-project/stream/stream-producer-8600/src/main/java/com/xjs/service/IMessageProducer.java

@@ -0,0 +1,14 @@
+package com.xjs.service;
+
+/**
+ * 消息生产者
+ *
+ * @author xiejs
+ * @since 2022-05-26
+ */
+public interface IMessageProducer {
+
+    void sendMessage();
+
+
+}

xjs-study/springcloud-project/stream/stream-producer-8600/src/main/java/com/xjs/service/impl/MessageProducerImpl.java

@@ -0,0 +1,31 @@
+package com.xjs.service.impl;
+
+import com.xjs.service.IMessageProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Service;
+
+import java.util.UUID;
+
+/**
+ * 消息生产者实现
+ *
+ * @author xiejs
+ * @since 2022-05-26
+ */
+@Service
+public class MessageProducerImpl implements IMessageProducer {
+
+    @Autowired
+    private StreamBridge streamBridge;
+
+
+    @Override
+    public void sendMessage() {
+        //向mq中发送消息(并不是直接操作mq，而是操作Spring Cloud Stream)
+        String message = UUID.randomUUID().toString();
+        streamBridge.send("myChannel-out-0", MessageBuilder.withPayload(message).build());
+
+    }
+}

xjs-study/springcloud-project/stream/stream-producer-8600/src/main/resources/application.yml

@@ -0,0 +1,35 @@
+server:
+  port: 8600
+
+spring:
+  application:
+    name: stream-producer
+
+  cloud:
+    stream:
+      binders:    #绑定mq服务信息
+        rabbitBinder:   #名称定义，用于后面的关联
+          type: rabbit
+          environment:    #mq环境配置
+            spring:
+              rabbitmq:
+                host: 192.168.23.128
+                port: 5672
+                username: guest
+                password: guest
+
+      bindings: # 关联整合通道和binder对象
+        myChannel-out-0:  # output是我们定义的通道名称，此处不能乱改
+          destination: exchange # 要使用的Exchange名称（消息队列主题名称）
+          content-type: application/json # application/json # 消息类型设置，比如json
+          binder: rabbitBinder # 关联MQ服务
+          group: default
+eureka:
+  client:
+    serviceUrl: # eureka server的路径
+      defaultZone: http://localhost:8761/eureka,http://localhost:8762/eureka,http://localhost:8763/eureka
+    instance:
+      prefer-ip-address: true #使用ip注册
+      instance-id: ${spring.cloud.client.ip-address}:${spring.application.name}:${server.port}   #实例名称
+
+

xjs-study/springcloud-project/stream/stream-producer-8600/src/test/java/com/xjs/service/impl/MessageProducerImplTest.java

@@ -0,0 +1,24 @@
+package com.xjs.service.impl;
+
+import com.xjs.StreamProducerApp8600;
+import com.xjs.service.IMessageProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * @author xiejs
+ * @since 2022-05-26
+ */
+@SpringBootTest(classes = StreamProducerApp8600.class)
+class MessageProducerImplTest {
+
+    @Autowired
+    private IMessageProducer iMessageProducer;
+
+    @org.junit.jupiter.api.Test
+    void sendMessage() {
+        iMessageProducer.sendMessage();
+    }
+}