zq-erp/src/main/java/com/matrix/config/RabbitMqConfig.java
@@ -36,7 +36,7 @@ public static final String MQ_EXCHANGE_A = "hive_exchange_A"; //订阅模式 public static final String MQ_EXCHANGE_FANOUT = "hive_exchange_fanout"; public static final String MQ_EXCHANGE_TOPIC = "hive_exchange_fanout"; @Bean VipCreateTask VipCreateTask() { @@ -46,6 +46,10 @@ @Bean OrderTask OrderrCreateTask() { return new OrderTask(); } @Bean SalesOrderTask SalesOrderTask() { return new SalesOrderTask(); } @Bean @@ -71,18 +75,24 @@ //声明一个交换机 rabiitMqTemplate.exchangeDeclare(MQ_EXCHANGE_A+evn, "direct"); rabiitMqTemplate.exchangeDeclare(MQ_EXCHANGE_FANOUT+evn, "fanout"); rabiitMqTemplate.exchangeDeclare(MQ_EXCHANGE_TOPIC +evn, "topic"); List<MqTask> taskList = new ArrayList<>(); //注册RabbitMq任务 taskList.add(new MqTask(MQ_EXCHANGE_A + evn, MQTaskRouting.CREATE_VIP + evn, MQTaskRouting.CREATE_VIP + evn, VipCreateTask())); //订阅模式 taskList.add(new MqTask(MQ_EXCHANGE_FANOUT + evn, MQTaskRouting.CREATE_ORDER + evn,MQTaskRouting.CREATE_ORDER + evn,OrderrCreateTask())); taskList.add(new MqTask(MQ_EXCHANGE_A + evn, MQTaskRouting.SEND_TEMPLATE_MSG + evn,MQTaskRouting.SEND_TEMPLATE_MSG + evn, TemplateMsgTask())); taskList.add(new MqTask(MQ_EXCHANGE_A + evn, MQTaskRouting.ORDER_OUT_SOTORE + evn,MQTaskRouting.ORDER_OUT_SOTORE + evn, OrderOutSotoreTask())); taskList.add(new MqTask(MQ_EXCHANGE_A + evn, MQTaskRouting.SEND_UNIFORM_TEMPLATE_MSG + evn,MQTaskRouting.SEND_UNIFORM_TEMPLATE_MSG + evn, UniformMsgSentTask())); //不同任务在不同的队列，但是routingKey一样则可以收到生产者消息 taskList.add(new MqTask(MQ_EXCHANGE_TOPIC + evn, MQTaskRouting.CREATE_ORDER + evn,MQTaskRouting.CREATE_ORDER + evn,SalesOrderTask())); taskList.add(new MqTask(MQ_EXCHANGE_TOPIC + evn, MQTaskRouting.SALES_ORDER + evn,MQTaskRouting.CREATE_ORDER + evn,OrderrCreateTask())); rabiitMqTemplate.binding(taskList); return rabiitMqTemplate; zq-erp/src/main/java/com/matrix/system/shopXcx/api/action/WxCommonAction.java
@@ -4,6 +4,7 @@ import com.matrix.component.wechat.externalInterface.common.Signature; import com.matrix.component.wechat.externalInterface.common.Util; import com.matrix.component.wechat.externalInterface.protocol.queryProtocol.NotifyData; import com.matrix.config.RabbitMqConfig; import com.matrix.core.pojo.AjaxResult; import com.matrix.core.tools.LogUtil; import com.matrix.system.common.bean.BusParameterSettings; @@ -192,7 +193,7 @@ } // 根据订单类型创建不同的处理任务 rabiitMqTemplate.sendMsg(MQTaskRouting.CREATE_ORDER+evn, orderId); rabiitMqTemplate.sendTopicMsg(RabbitMqConfig.MQ_EXCHANGE_TOPIC +evn, MQTaskRouting.CREATE_ORDER+evn, orderId); threadResult.putInMap("status", "success"); zq-erp/src/main/java/com/matrix/system/shopXcx/mqTask/MQTaskRouting.java
@@ -8,6 +8,7 @@ String CREATE_VIP = "CREATE_VIP"; String CREATE_ORDER = "CREATE_ORDER"; String SALES_ORDER = "SALES_ORDER"; /** * 发送微信消息 */ zq-erp/src/test/java/com/matrix/SourceFlowTests.java
@@ -1,11 +1,14 @@ package com.matrix; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.matrix.component.rabbitmq.RabiitMqTemplate; import com.matrix.config.RabbitMqConfig; import com.matrix.core.tools.StringUtils; import com.matrix.system.fenxiao.dao.ShopSalesmanApplyDao; import com.matrix.system.hive.bean.*; import com.matrix.system.hive.dao.*; import com.matrix.system.hive.service.CodeService; import com.matrix.system.shopXcx.mqTask.MQTaskRouting; import com.matrix.system.shopXcx.vo.SalesmanVo; import org.junit.Test; import org.junit.runner.RunWith; @@ -49,8 +52,20 @@ @Autowired private SysVipInfoDao vipInfoDao; @Autowired RabiitMqTemplate rabiitMqTemplate; /** * 发送订阅消息 * @throws InterruptedException */ @Test public void testTopic() throws InterruptedException { rabiitMqTemplate.sendTopicMsg(RabbitMqConfig.MQ_EXCHANGE_TOPIC +"dev", MQTaskRouting.CREATE_ORDER+"dev", "123"); Thread.sleep(10000); } @Test @Transactional @Rollback
