消息驱动_入门案例之消息生产者-【官方】百战程序员_IT在线教育培训机构


xxxxxxxxxx
 <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>

或者


xxxxxxxxxx
<dependency>
     <groupId>org.springframework.cloud</groupId>
     <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

完整依赖如下：


xxxxxxxxxx
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <!--  引入Eureka client依赖   -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.22</version>
        </dependency>
    </dependencies>

编写YML文件


xxxxxxxxxx
server:
  port: 800
eureka:
  instance:
    # 实例名字
    instance-id: stream-producer8001
  client:
    service-url:
      # 指定单机eureka server地址
      #defaultZone: http://localhost:7001/eureka/
      defaultZone: http://localhost:7001/eureka/,http://localhost:7002/eureka/
spring:
  application:
    name: stream-producer
  rabbitmq:
    port: 5672
    host: 192.168.66.101
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        # 广播消息
        myBroadcast-out-0: # 生产者绑定名称，myBroadcast是自定义的绑定名称，out代表生产者，0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange

注意：
spring.cloud.function.definition 属性非常重要，务必配置。

消息实体类


xxxxxxxxxx
@Setter
@Getter
public class MyMessage implements java.io.Serializable {

    // 消息体
    private String payload;
}

消息生产类实现


xxxxxxxxxx
   package com.itbaizhan.service.impl;

import com.itbaizhan.service.IMessageProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Service;

/**
 * 定义消息的推送管道
 */
@Service
public class MessageProviderImpl implements IMessageProvider {

    @Autowired
    private StreamBridge streamBridge;

    @Override
    public String send(String message) {

        MyMessage myMessage = new MyMessage();
        myMessage.setPayload(message);

// 生产消息
        // 第一个参数是绑定名称，格式为：自定义的绑定名称-out-0，myBroadcast是自定义的绑定名称，out代表生产者，0是固定写法
        // 自定义的绑定名称必须与消费方法的方法名保持一致
        // 第二个参数是发送的消息实体
        streamBridge.send("myBroadcast-out-0", myMessage);
        return "SUCCESS";
    }
}

编写业务类


xxxxxxxxxx
@RestController
public class SendMessageController {

    @Resource
    private IMessageProvider messageProvider;

    @GetMapping("/send")
    public String sendMessage(String message){
        return messageProvider.send(message);
    }

}

测试

自动生成队列

消息驱动_什么是Spring Cloud Stream 消息驱动_入门案例之消息消费者

北京市昌平区回龙观镇南店村综合商业楼2楼226室