文章目錄

• eureka作注冊中心的配置：
• 消息提供方：
• 消費者代碼

注冊中心、消息接受者、消息提供者分別啟動：

eureka作注冊中心的配置：

依賴包：

<dependencies><!--eureka-server--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-server</artifactId></dependency><!-- 引入自己定義的api通用包，可以使用Payment支付Entity --><dependency><groupId>com.atguigu.springcloud</groupId><artifactId>cloud-api-commons</artifactId><version>${project.version}</version></dependency><!--boot web actuator--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><!--一般通用配置--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId></dependency></dependencies> server:port: 7001eureka:instance:hostname: eureka7001.com #eureka服務端的實例名稱client:register-with-eureka: false #false表示不向注冊中心注冊自己。fetch-registry: false #false表示自己端就是注冊中心，我的職責就是維護服務實例，并不需要去檢索服務service-url:#集群指向其它eureka#defaultZone: http://eureka7002.com:7002/eureka/#單機就是7001自己defaultZone: http://eureka7001.com:7001/eureka/#server:#關閉自我保護機制，保證不可用服務被及時踢除#enable-self-preservation: false#eviction-interval-timer-in-ms: 2000

消息提供方：

目錄：

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><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-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

配置：

server:port: 8801spring:application:name: cloud-stream-providercloud:stream:binders: # 在此處配置要綁定的rabbitmq的服務信息；defaultRabbit: # 表示定義的名稱，用于于binding整合type: rabbit # 消息組件類型environment: # 設置rabbitmq的相關的環境配置spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服務的整合處理output: # 這個名字是一個通道的名稱destination: studyExchange # 表示要使用的Exchange名稱定義content-type: application/json # 設置消息類型，本次為json，文本則設置“text/plain”binder: defaultRabbit # 設置要綁定的消息服務的具體設置eureka:client: # 客戶端進行Eureka注冊的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 設置心跳的時間間隔（默認是30秒）lease-expiration-duration-in-seconds: 5 # 如果現在超過了5秒的間隔（默認是90秒）instance-id: send-8801.com # 在信息列表時顯示主機名稱prefer-ip-address: true # 訪問的路徑變為IP地址

接口：

public interface IMessageProvider {public String send(); } @EnableBinding(Source.class) //定義消息的推送管道 public class MessageProviderImpl implements IMessageProvider {@Resourceprivate MessageChannel output; // 消息發送管道@Overridepublic String send(){String serial = UUID.randomUUID().toString();output.send(MessageBuilder.withPayload(serial).build());System.out.println("*****serial: "+serial);return null;} }

controller:

@RestController public class SendMessageController {@Resourceprivate IMessageProvider messageProvider;@GetMapping(value = "/sendMessage")public String sendMessage(){return messageProvider.send();}}

消費者代碼

配置和目錄與消息提供方幾乎一致

server:port: 8803spring:application:name: cloud-stream-consumercloud:stream:binders: # 在此處配置要綁定的rabbitmq的服務信息；defaultRabbit: # 表示定義的名稱，用于于binding整合type: rabbit # 消息組件類型environment: # 設置rabbitmq的相關的環境配置spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服務的整合處理input: # 這個名字是一個通道的名稱destination: studyExchange # 表示要使用的Exchange名稱定義content-type: application/json # 設置消息類型，本次為對象json，如果是文本則設置“text/plain”binder: defaultRabbit # 設置要綁定的消息服務的具體設置group: atguiguAeureka:client: # 客戶端進行Eureka注冊的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 設置心跳的時間間隔（默認是30秒）lease-expiration-duration-in-seconds: 5 # 如果現在超過了5秒的間隔（默認是90秒）instance-id: receive-8803.com # 在信息列表時顯示主機名稱prefer-ip-address: true # 訪問的路徑變為IP地址

controller：

@Component @EnableBinding(Sink.class) //@EnableBinding 對應的兩個接口屬性 Source 和 Sink 是 SCS 內部提供的。 // SCS 內部會基于 Source 和 Sink 構造 BindableProxyFactory， // 且對應的 output 和 input 方法返回的 MessageChannel 是 DirectChannel。output 和 input 方法修飾的注解對應的 value 是配置文件中 binding 的 name。 //————————————————public class ReceiveMessageListenerController {@Value("${server.port}")private String serverPort;@StreamListener(Sink.INPUT)public void input(Message<String> message){System.out.println("消費者2號,----->接受到的消息: "+message.getPayload()+"\t port: "+serverPort);} }

訪問消息提供者的接口http://localhost:8801/sendMessage
控制臺輸出：

總結

以上是生活随笔為你收集整理的springcloudstream+rabbitmq+eureka进行消息发送和接收实例代码的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。