diff --git a/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java b/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java index 47de9b6..fc861d7 100644 --- a/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java +++ b/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java @@ -223,7 +223,7 @@ public class AgentEndpoint { jsonObject.put("dataType", MsgEnum.网络上报重试.getValue()); jsonObject.put("data", data); MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build(); - agentProducer.asyncSend(mqMsg); + agentProducer.sendOrderly(mqMsg, clientId); JSONObject json = new JSONObject(); json.put("resCode",1); return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.采集上报应答.getValue()) diff --git a/src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java b/src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java index 4531f53..f34235a 100644 --- a/src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java +++ b/src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java @@ -51,4 +51,19 @@ public class AgentProducer implements RocketMqService { rocketMQTemplate.syncSend(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build(), 30000, msg.getLevel()); } + // 新增顺序发送 + @Override + public void sendOrderly(MqMsg mqMsg, String shardingKey) { + try { + AssertLog.info("sendOrderly发送消息:==>{}", mqMsg); + rocketMQTemplate.syncSendOrderly( + mqMsg.getTopic(), + MessageBuilder.withPayload(mqMsg.getContent()).build(), + shardingKey + ); + } catch (Exception e) { + AssertLog.error("顺序发送消息失败", e); + } + } + } diff --git a/src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java b/src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java index 6cb5894..f6e6cf4 100644 --- a/src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java +++ b/src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java @@ -41,5 +41,12 @@ public interface RocketMqService { * @param msg 发送消息实体类 */ default void delayedSendOrderly(MqMsg msg){}; + + /** + * 发送顺序消息 + * @param mqMsg + * @param shardingKey + */ + public void sendOrderly(MqMsg mqMsg, String shardingKey); } \ No newline at end of file