SpringCloud消息驱动Stream实例

2022-07-04

依赖

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

配置yml

spring:
  application:
    name: folkspace-market-api
  profiles:
    active: default
  rabbitmq:
    host: ${folk_rabbitmq_host}
    port: ${folk_rabbitmq_port}
    username: ${folk_rabbitmq_username}
    password: ${folk_rabbitmq_password}
    virtual-host: ${folk_rabbitmq_virtual_host}
  cloud:
    loadbalancer:
      ribbon:
        enabled: false
    stream:
      default-binder: rabbit
      rabbit:
        bindings:
          # 单播
          input:
            consumer:
              delayedExchange: true
              bindingRoutingKeyDelimiter: ','
              bindingRoutingKey: OrderMessage
          output:
            producer:
              delayedExchange: true
              routingKeyExpression: headers.type
              autoBindDlq: true
          # 广播
          repInput:
            consumer:
              delayedExchange: true
              bindingRoutingKeyDelimiter: ','
              bindingRoutingKey: MarketNftMessage
          repOutput:
            producer:
              delayedExchange: true
              routingKeyExpression: headers.type
              autoBindDlq: true
      bindings:
        # 单播
        input:
          group: ${spring.application.name}
          destination: folkspace-app-default
          binder: rabbit
        output:
          destination: folkspace-app-default
          binder: rabbit
        # 广播
        repInput:
          destination: folkspace-app-repository
          binder: rabbit
        repOutput:
          destination: folkspace-app-repository
          binder: rabbit

配置properties

# port
server.port=8002
# rabbitMq
folk_rabbitmq_host=xxx
folk_rabbitmq_port=5672
folk_rabbitmq_username=xxx
folk_rabbitmq_password=xxx
folk_rabbitmq_virtual_host=folkspace-app-prod

广播

发送消息服务类

import com.folkspace.base.config.RepOutput;
import com.folkspace.demo.contact.message.GoodsHmMessage;
import com.folkspace.demo.contact.message.GoodsMessage;
import com.folkspace.demo.contact.message.MarketHmNftMessage;
import com.folkspace.demo.contact.message.MarketNftMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
public class MarketSendMq {
  
    @Autowired
    private RepOutput repOutput;

    /**
     * 计算熔断级别、计算指导价、更新nft_status
     */
    public void sendMarketNftMessage(String nftTmpId, int flag) {
        new MarketNftMessage()
                .setNftTmpId(nftTmpId)
                .setFlag(flag)
                .send(repOutput);
    }

}

MarketNftMessage

import com.folkspace.base.message.BaseMessage;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

@Setter
@Getter
@Accessors(chain = true)
public class MarketNftMessage extends BaseMessage {
  
    private int flag;
    private String nftTmpId;

    @AllArgsConstructor
    @Getter
    public enum MarketNftMessageEnum {
        UPDATE_PRICE(1, "修改NFT价格"),
        UPDATE_SALE(2, "修改开售状态"),
        UPDATE_STATUS(3, "修改NFT状态"),
        ;
        private final int value;
        private final String text;
    }
}

RepOutput

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

public interface RepOutput {
    String OUTPUT = "repOutput";

    @Output(RepOutput.OUTPUT)
    MessageChannel output();
}

BaseMessage

import com.alibaba.fastjson.JSON;
import com.folkspace.base.config.RepOutput;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;

import java.io.Serializable;

@Slf4j
@Data
@Accessors(chain = true)
public class BaseMessage implements Serializable {

    public boolean send(Source source) {
        String type = this.getClass().getSimpleName();
        log.warn("SEND Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
        return source.output().send(MessageBuilder
                .withPayload(this)
                .setHeader("type", type)
                .build());
    }

    public boolean sendDelay(Source source, long delaySecond) {
        String type = this.getClass().getSimpleName();
        log.warn("SEND Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
        return source.output().send(MessageBuilder
                .withPayload(this)
                .setHeader("type", type)
                .setHeader("x-delay", delaySecond)
                .build());
    }

    public void recv(Logger log) {
        log.info("RECV Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
    }

    public boolean send(RepOutput repOutput) {
        String type = this.getClass().getSimpleName();
        log.warn("SEND Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
        return repOutput.output().send(MessageBuilder
                .withPayload(this)
                .setHeader("type", type)
                .build());
    }

    public boolean sendDelay(RepOutput repOutput, long delaySecond) {
        String type = this.getClass().getSimpleName();
        log.warn("SEND Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
        return repOutput.output().send(MessageBuilder
                .withPayload(this)
                .setHeader("type", type)
                .setHeader("x-delay", delaySecond)
                .build());
    }
}

接收消息服务类


@Slf4j
@Component
public class MarketConsumerMq {

    // 消费：熔断时、计算等消息
    @StreamListener(value = RepInput.INPUT, condition = "headers['type'] == 'MarketNftMessage'")
    public void consumerRepMarketNftMessage(MarketNftMessage message) {
        log.info("consumerRepMarketNftMessage={}", JSON.toJSONString(message));
      
        // 消费内容
    }
}

RepInput

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

public interface RepInput {
    String INPUT = "repInput";

    @Input("repInput")
    SubscribableChannel input();
}

单播

发送消息服务类

import com.folkspace.order.api.constants.OrderConstants;
import com.folkspace.order.contact.message.OrderMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class OrderSendMq {
    
    @Autowired
    private Source source;

    public void sendOrderPaySuccessMessage(OrderMessage message, long delaySecond) {
        message.sendDelay(source, delaySecond);
    }

    public void sendOrderExpiryMessage(OrderMessage message) {
        message.sendDelay(source, OrderConstants.ORDER_EXPIRY_TIME);
    }
}

OrderMessage

import com.folkspace.base.message.BaseMessage;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

@Setter
@Getter
@Accessors(chain = true)
public class OrderMessage extends BaseMessage {
    @Schema(description = "订单号")
    private String orderId;

    @Schema(description = "订单消息类型 ORDER_EXPIRY：订单失效 PAY_SUCCESS：付款成功 DATA_UPDATE：nft、goods熔断更新")
    private String messageType;
}

BaseMessage

import com.alibaba.fastjson.JSON;
import java.io.Serializable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;

public class BaseMessage implements Serializable {
    private static final Logger log = LoggerFactory.getLogger(BaseMessage.class);

    public boolean send(Source source) {
        String type = this.getClass().getSimpleName();
        log.warn("SEND Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
        return source.output().send(MessageBuilder.withPayload(this).setHeader("type", type).build());
    }

    public boolean sendDelay(Source source, long delaySecond) {
        String type = this.getClass().getSimpleName();
        log.warn("SEND Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
        return source.output().send(MessageBuilder.withPayload(this).setHeader("type", type).setHeader("x-delay", delaySecond).build());
    }

    public void recv(Logger log) {
        log.info("RECV Message: {},{}", this.getClass().getSimpleName(), JSON.toJSONString(this));
    }

    public BaseMessage() {
    }

    public boolean equals(final Object o) {
        if (o == this) {
            return true;
        } else if (!(o instanceof BaseMessage)) {
            return false;
        } else {
            BaseMessage other = (BaseMessage)o;
            return other.canEqual(this);
        }
    }

    protected boolean canEqual(final Object other) {
        return other instanceof BaseMessage;
    }

    public int hashCode() {
        int result = true;
        return 1;
    }

    public String toString() {
        return "BaseMessage()";
    }
}

接收消息服务类

import com.alibaba.fastjson.JSON;
import com.folkspace.order.api.service.OrderService;
import com.folkspace.order.contact.message.OrderMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class OrderConsumerMq {

    @Autowired
    private OrderService orderService;

    @StreamListener(value = Sink.INPUT, condition = "headers['type'] == 'OrderMessage'")
    public void consumerOrderMessage(OrderMessage message) {
        log.info("consumerOrderMessage={}", JSON.toJSONString(message));
        switch (message.getMessageType()) {
            case "ORDER_EXPIRY":
                orderService.payExpiry(message.getOrderId());
                break;
            case "PAY_SUCCESS":
                orderService.paySuccess(message.getOrderId());
                break;
            case "DATA_UPDATE":

                break;
            default:
                log.error("订单消息消费失败");
                break;
        }
    }
}