coinex06 // 前端数据 -＞ ringbuffer -＞ cpu

目录

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 构建完成 // 构建笔记https://blog.csdn.net/tenc1239/article/details/130162335?spm=1001.2014.3001.55022 exchange-service 前端操作数据 -> RingBuffer , Disraptor 方法-> match-service3 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 sentbinders: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())); // 交易sideorder.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 { // 消息消费者@Autowiredprivate 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 队列}
}