目录

0. 逻辑树

1 exchange-service 发送消息

1.1 exchange-service 添加依赖

1.2. yml配置文件

1.3. Source

1.4. 配置类

1.5. 发送消息到撮合引擎  service -> impl -> EntrustOrderServiceImpl

1.6. recket-server:8080 

2. match-server 接收数据

2.1 数据转换

2.2 监听接收数据 -> 发送到ringbuffer

---

0. 逻辑树

1 RingBuffer , Disraptor 构建完成

2 exchange-service 前端操作数据 -> RingBuffer , Disraptor 方法-> match-service

3 match-service 接收订单 -> 撮合引擎

1 exchange-service 发送消息

1.1 exchange-service 添加依赖

 <!--        spring-cloud-stream-rocketmq-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
        </dependency>

1.2. yml配置文件

    stream: # 交易微服务里的数据发送
      rocketmq:
        binder:
          name-server: rocket-server:9876  #/RocketMQ Message hasn't been sent
        binders:
          order_out: {consumer.orderly: true}
      bindings:
        order_out: {destination: order_in, content-type: application/plain}
       

1.3. Source

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

public interface Source {

    @Output("order_out") // 要在 impl  EntrustOrderServiceImpl中 远程发送 到撮合引擎
    MessageChannel outputMessage() ; //导包 import org.springframework.messaging.MessageChannel;
    
}

1.4. 配置类

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

@Configuration
@EnableBinding(value = {Sink.class , Source.class})
public class RocketMQConfig { // 配置类
}

1.5. 发送消息到撮合引擎  service -> impl -> EntrustOrderServiceImpl

MessageBuilder<EntrustOrder> entrustOrderMessageBuilder = MessageBuilder.withPayload(entrustOrder).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
     
source.outputMessage().send(entrustOrderMessageBuilder.build()); // 点击买入 卖出按钮后 执行 发送到撮合交易系统

1.6. recket-server:8080 

2. match-server 接收数据

2.1 数据转换

import org.example.domain.EntrustOrder;
import org.example.enums.OrderDirection;
import org.example.model.Order;

public class BeanUtils {


    /**
     * 将EntrustOrder 转化为我们的Order
     * @param entrustOrder
     * @return
     */
    public static  Order entrustOrder2Order(EntrustOrder entrustOrder) {
        Order order = new Order();
        order.setOrderId(entrustOrder.getId().toString());

        order.setPrice(entrustOrder.getPrice());
        order.setAmount(entrustOrder.getVolume().subtract(entrustOrder.getDeal())); // 交易的数量= 总数量- 已经成交的数量

        order.setSymbol(entrustOrder.getSymbol());
        order.setOrderDirection(OrderDirection.getOrderDirection(entrustOrder.getType().intValue())); // 交易side
        order.setTime(entrustOrder.getCreated().getTime());

        return order ;

    }
}

2.2 监听接收数据 -> 发送到ringbuffer

match-server -> rocket -> MessageConsumerListener

import org.example.disruptor.DisruptorTemplate;

import org.example.domain.EntrustOrder;
import org.example.model.Order;

import lombok.extern.slf4j.Slf4j;
import org.example.util.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class MessageConsumerListener { // 消息消费者

    @Autowired
    private DisruptorTemplate disruptorTemplate;

    @StreamListener("order_in") // "order_in" 在 Sink中
    public void handleMessage(EntrustOrder entrustOrder) { // 消息监听
        Order order = null;
        if (entrustOrder.getStatus() == 2) { // 该单需要取消
            order = new Order();
            order.setOrderId(entrustOrder.getId().toString());
            order.setCancelOrder(true);
        } else {
            order = BeanUtils.entrustOrder2Order(entrustOrder);
        }

        log.info("接收到了委托单:{}", order);
        disruptorTemplate.onData(order);// 将order 发送到ringbuf 队列
    }
}