project-zsy
/
gasWellWatch
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

设备数据持久化事件推送队列动态生成

This commit is contained in:
wangshilong 2024-11-27 15:06:43 +08:00
parent a7f4dcd000
commit 5b7c3a10f9
2 changed files with 19 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/java/com/isu/gaswellwatch/modbus/data/listener/ComposeModbusMessageListener.java
View File

@ -29,4 +29,5 @@ public class ComposeModbusMessageListener implements BatchMessageListener {
public void onMessageBatch(List<Message> messages) { public void onMessageBatch(List<Message> messages) {
this.messageListenerList.forEach(messageListener -> messageListener.onMessageBatch(messages)); this.messageListenerList.forEach(messageListener -> messageListener.onMessageBatch(messages));
} }
} }

45
src/main/java/com/isu/gaswellwatch/modbus/data/listener/DynamicRabbitListener.java
View File

@ -28,40 +28,30 @@ public class DynamicRabbitListener implements ApplicationRunner {
private final AmqpAdmin amqpAdmin; private final AmqpAdmin amqpAdmin;
private final JdbcTemplate jdbcTemplate; private final JdbcTemplate jdbcTemplate;
private final ComposeModbusMessageListener composeListener; private final ComposeModbusMessageListener composeListener;
private final SimpleMessageListenerContainer listenerContainer; private final SimpleMessageListenerContainer modbusMessageListenerContainer;
public DynamicRabbitListener(ConnectionFactory connectionFactory, JdbcTemplate jdbcTemplate) { public DynamicRabbitListener(ConnectionFactory connectionFactory, JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate; this.jdbcTemplate = jdbcTemplate;
this.amqpAdmin = new RabbitAdmin(connectionFactory); this.amqpAdmin = new RabbitAdmin(connectionFactory);
this.listenerContainer = new SimpleMessageListenerContainer(); this.modbusMessageListenerContainer = new SimpleMessageListenerContainer();
this.listenerContainer.setConnectionFactory(connectionFactory); this.modbusMessageListenerContainer.setConnectionFactory(connectionFactory);
this.listenerContainer.setMessageListener(this.composeListener = new ComposeModbusMessageListener()); this.modbusMessageListenerContainer.setMessageListener(this.composeListener = new ComposeModbusMessageListener());
// 启动监听容器 // 启动监听容器
this.listenerContainer.start(); this.modbusMessageListenerContainer.start();
} }
@Override @Override
public void run(ApplicationArguments args) { public void run(ApplicationArguments args) {
this.composeListener.addBatchMessageListener(new ModbusMessageBackupListener()); this.composeListener.addBatchMessageListener(new ModbusMessageBackupListener());
this.composeListener.addBatchMessageListener(new ModbusMessagePersistListener()); this.composeListener.addBatchMessageListener(new ModbusMessagePersistListener());
this.jdbcTemplate.queryForList(GATEWAY_SQL).forEach(this::registerDeviceAndListener);
this.jdbcTemplate.queryForList(GATEWAY_SQL).forEach(item ->
this.addListenerQueue(this.registerDevice(item), null, null));
} }
public void registerDevice(Long deviceId) { public void registerDeviceAndListener(Long deviceId) {
this.jdbcTemplate.queryForList(DEVICE_SQL + deviceId).forEach(this::registerDevice); this.jdbcTemplate.queryForList(DEVICE_SQL + deviceId).forEach(this::registerDeviceAndListener);
} }
public void addListenerQueue(String gatewaySn) { public void addListenerCollectQueue(Queue queue, String exchangeName, String routingKey) {
this.addListenerQueue(String.format(Queues.MODBUS_COLLECT_DATA, gatewaySn), null, null);
}
public void addListenerQueue(String queueName, String exchangeName, String routingKey) {
Queue queue = QueueBuilder.durable(queueName).build();
// 声明队列
this.amqpAdmin.declareQueue(queue);
if (StringUtils.isNotBlank(exchangeName) && StringUtils.isNotBlank(routingKey)) { if (StringUtils.isNotBlank(exchangeName) && StringUtils.isNotBlank(routingKey)) {
// 声明直接类型的交换器 // 声明直接类型的交换器
DirectExchange exchange = new DirectExchange(exchangeName); DirectExchange exchange = new DirectExchange(exchangeName);
@ -69,17 +59,18 @@ public class DynamicRabbitListener implements ApplicationRunner {
// 绑定队列和交换器 // 绑定队列和交换器
this.amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey)); this.amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey));
} }
// 设置监听的队列 // 设置监听的队列
this.listenerContainer.addQueues(queue); this.modbusMessageListenerContainer.addQueues(queue);
} }
private String registerDevice(Map<String, Object> deviceMap) { private void registerDeviceAndListener(Map<String, Object> deviceMap) {
String eventQueue = String.format(Queues.MODBUS_COLLECT_DATA, MapUtil.getStr(deviceMap, "id")); Queue eventQueue = QueueBuilder.durable(String.format(Queues.DEVICE_EVENTS,
String collectDataQueue = String.format(Queues.MODBUS_COLLECT_DATA, MapUtil.getStr(deviceMap, "identifier")); MapUtil.getStr(deviceMap, "id"))).build();
this.amqpAdmin.declareQueue(QueueBuilder.durable(eventQueue).build()); Queue collectQueue = QueueBuilder.durable(String.format(Queues.MODBUS_COLLECT_DATA,
this.amqpAdmin.declareQueue(QueueBuilder.durable(collectDataQueue).build()); MapUtil.getStr(deviceMap, "identifier"))).build();
return collectDataQueue; this.amqpAdmin.declareQueue(eventQueue);
this.amqpAdmin.declareQueue(collectQueue);
this.addListenerCollectQueue(collectQueue, null, null);
} }
} }