（十七）JAVA springcloud ssm b2b2c多用户商城系统-消息驱动 Spring Cloud Stream

在使用spring cloud云架构的时候，我们不得不使用Spring cloud Stream，因为消息中间件的使用在项目中无处不在，我们公司后面做了娱乐方面的APP，在使用spring cloud做架构的时候，其中消息的异步通知，业务的异步处理都需要使用消息中间件机制。spring cloud的官方给出的集成建议（使用rabbit mq和kafka），我看了一下源码和配置，只要把rabbit mq集成，kafka只是换了一个pom配置jar包而已，闲话少说，我们就直接进入配置实施：

1. 简介：

Spring cloud Stream 数据流操作开发包，封装了与Redis,Rabbit、Kafka等发送接收消息。

2. 使用工具：

rabbit，具体的下载和安装细节我这里不做太多讲解，网上的实例太多了

3. 创建commonservice-mq-producer消息的发送者项目，在pom里面配置stream-rabbit的依赖

<span style="font-size: 16px;"><!-- 引入MQ消息驱动的微服务包，引入stream只需要进行配置化即可，是对rabbit、kafka很好的封装 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency></span>

4. 在yml文件里面配置rabbit mq

<span style="font-size: 16px;">server:
  port: 5666
spring:
  application:
    name: commonservice-mq-producer
  profiles:
    active: dev
  cloud:
    config:
      discovery:
        enabled: true
        service-id: commonservice-config-server
  <span style="color: #ff0000;"># rabbitmq和kafka都有相关配置的默认值，如果修改，可以再次进行配置
    stream:
      bindings:
        mqScoreOutput:
          destination: honghu_exchange
          contentType: application/json  

  rabbitmq:
     host: localhost
     port: 5672
     username: honghu
     password: honghu</span>
eureka:
  client:
    service-url:
      defaultZone: http://honghu:[email protected]:8761/eureka
  instance:
    prefer-ip-address: true</span>

5. 定义接口ProducerService

<span style="font-size: 16px;">package com.honghu.cloud.producer;  

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.SubscribableChannel;  

public interface ProducerService {  

    String SCORE_OUPUT = "mqScoreOutput";  

    @Output(ProducerService.SCORE_OUPUT)
    SubscribableChannel sendMessage();
}</span>

6. 定义绑定

<span style="font-size: 16px;">package com.honghu.cloud.producer;  

import org.springframework.cloud.stream.annotation.EnableBinding;  

@EnableBinding(ProducerService.class)
public class SendServerConfig {  

}</span>

7. 定义发送消息业务ProducerController

<span style="font-size: 16px;">package com.honghu.cloud.controller;  

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;  

import com.honghu.cloud.common.code.ResponseCode;
import com.honghu.cloud.common.code.ResponseVO;
import com.honghu.cloud.entity.User;
import com.honghu.cloud.producer.ProducerService;  

import net.sf.json.JSONObject;  

@RestController
@RequestMapping(value = "producer")
public class ProducerController {  

    @Autowired
    private ProducerService producerService;  

    /**
     * 通过get方式发送</span>对象<span style="font-size: 16px;">
     * @param name 路径参数
     * @return 成功|失败
     */
    @RequestMapping(value = "/sendObj", method = RequestMethod.GET)
    public ResponseVO sendObj() {
        User user = new User(1, "hello User");
        <span style="color: #ff0000;">Message<User> msg = MessageBuilder.withPayload(user).build();</span>
        boolean result = producerService.sendMessage().send(msg);
        if(result){
            return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false);
        }
        return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false);
    }  

    /**
     * 通过get方式发送字符串消息
     * @param name 路径参数
     * @return 成功|失败
     */
    @RequestMapping(value = "/send/{name}", method = RequestMethod.GET)
    public ResponseVO send(@PathVariable(value = "name", required = true) String name) {
        Message msg = MessageBuilder.withPayload(name.getBytes()).build();
        boolean result = producerService.sendMessage().send(msg);
        if(result){
            return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false);
        }
        return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false);
    }  

    /**
     * 通过post方式发送</span>json对象<span style="font-size: 16px;">
     * @param name 路径参数
     * @return 成功|失败
     */
    @RequestMapping(value = "/sendJsonObj", method = RequestMethod.POST)
    public ResponseVO sendJsonObj(@RequestBody JSONObject jsonObj) {
        Message<JSONObject> msg = MessageBuilder.withPayload(jsonObj).build();
        boolean result = producerService.sendMessage().send(msg);
        if(result){
            return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false);
        }
        return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false);
    }
}
</span>

8. 创建commonservice-mq-consumer1消息的消费者项目，在pom里面配置stream-rabbit的依赖

<!-- 引入MQ消息驱动的微服务包，引入stream只需要进行配置化即可，是对rabbit、kafka很好的封装 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

9. 在yml文件中配置：

server:
  port: 5111
spring:
  application:
    name: commonservice-mq-consumer1
  profiles:
    active: dev
  cloud:
    config:
      discovery:
        enabled: true
        service-id: commonservice-config-server  

    <span style="color: #ff0000;">stream:
      bindings:
        mqScoreInput:
          group: honghu_queue
          destination: honghu_exchange
          contentType: application/json  

  rabbitmq:
     host: localhost
     port: 5672
     username: honghu
     password: honghu</span>
eureka:
  client:
    service-url:
      defaultZone: http://honghu:[email protected]:8761/eureka
  instance:
    prefer-ip-address: true

10. 定义接口ConsumerService

package com.honghu.cloud.consumer;  

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;  

public interface ConsumerService {  

    <span style="color: #ff0000;">String SCORE_INPUT = "mqScoreInput";  

    @Input(ConsumerService.SCORE_INPUT)
    SubscribableChannel sendMessage();</span>  

}

11. 定义启动类和消息消费

package com.honghu.cloud;  

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;  

import com.honghu.cloud.consumer.ConsumerService;
import com.honghu.cloud.entity.User;  

@EnableEurekaClient
@SpringBootApplication
@EnableBinding(ConsumerService.class) //可以绑定多个接口
public class ConsumerApplication {  

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

    <span style="color: #ff0000;">@StreamListener(ConsumerService.SCORE_INPUT)
    public void onMessage(Object obj) {
        System.out.println("消费者1，接收到的消息：" + obj);
    }</span>  

}

12. 分别启动commonservice-mq-producer、commonservice-mq-consumer1

13. 通过postman来验证消息的发送和接收

可以看到接收到了消息，下一章我们介绍mq的集群方案。

到此，整个消息中心方案集成完毕

欢迎大家和我一起学习spring cloud构建微服务云架构，我这边会将近期研发的spring cloud微服务云架构的搭建过程和精髓记录下来，帮助更多有兴趣研发spring cloud框架的朋友，大家来一起探讨spring cloud架构的搭建过程及如何运用于企业项目。
需要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码请加企鹅求求：贰一四七七七五六叁叁完整项目的源码来源

原文地址：https://www.cnblogs.com/sccoming/p/10141634.html

时间： 2024-10-09 00:14:22

（十七）JAVA springcloud ssm b2b2c多用户商城系统-消息驱动 Spring Cloud Stream

（十七）JAVA springcloud ssm b2b2c多用户商城系统-消息驱动 Spring Cloud Stream的相关文章

（十六）JAVA springcloud ssm b2b2c多用户商城系统-使用spring cloud Bus刷新配置

JAVA springcloud ssm b2b2c多用户商城系统源码-docker-feign-hystrix（六）

(三)JAVA springcloud ssm b2b2c多用户商城系统：服务提供与调用

JAVA springboot ssm b2b2c多用户商城系统源码（六）分布式配置中心(Spring Cloud Config)

SpringBoot b2b2c 多用户商城系统-docker-feign配置(五)

SpringBoot b2b2c 多用户商城系统（十五）Springboot整合RabbitMQ

SpringBoot b2b2c 多用户商城系统（十六）用restTemplate消费服务

Java B2B2C多用户商城 springboot架构-config-bus(十三)

（十二）Java B2B2C多用户商城 springboot架构-SSO单点登录之OAuth2.0