| @@ -1,10 +1,12 @@ | |||||
| package com.iformall.mq.impl; | package com.iformall.mq.impl; | ||||
| import com.iformall.mq.MqBaseConsumer; | import com.iformall.mq.MqBaseConsumer; | ||||
| import com.iformall.enums.EnumMsgMqTopic; | |||||
| import com.iformall.mq.MQConfig; | import com.iformall.mq.MQConfig; | ||||
| import org.springframework.amqp.core.*; | import org.springframework.amqp.core.*; | ||||
| import org.springframework.amqp.rabbit.annotation.RabbitListener; | import org.springframework.amqp.rabbit.annotation.RabbitListener; | ||||
| import org.springframework.beans.factory.annotation.Autowired; | import org.springframework.beans.factory.annotation.Autowired; | ||||
| import org.springframework.context.annotation.Bean; | |||||
| import org.springframework.context.annotation.Profile; | import org.springframework.context.annotation.Profile; | ||||
| import org.springframework.stereotype.Component; | import org.springframework.stereotype.Component; | ||||
| @@ -13,29 +15,42 @@ import javax.annotation.PostConstruct; | |||||
| @Component | @Component | ||||
| @Profile(MQConfig.Impl.RABBIT_MQ) | @Profile(MQConfig.Impl.RABBIT_MQ) | ||||
| public class RabbitMqConsumer extends MqBaseConsumer { | public class RabbitMqConsumer extends MqBaseConsumer { | ||||
| String queueName = "topic-1"; | |||||
| //String queueName = "topic-1"; | |||||
| String queuenames = ""; | |||||
| @Autowired | @Autowired | ||||
| private AmqpAdmin amqpAdmin; | private AmqpAdmin amqpAdmin; | ||||
| @PostConstruct | @PostConstruct | ||||
| public void init() { | public void init() { | ||||
| DirectExchange exchange = new DirectExchange(queueName); | |||||
| Queue queue = new Queue(queueName); | |||||
| Binding binding = BindingBuilder.bind(queue).to(exchange).withQueueName(); | |||||
| amqpAdmin.declareExchange(exchange); | |||||
| amqpAdmin.declareQueue(queue); | |||||
| amqpAdmin.declareBinding(binding); | |||||
| for (EnumMsgMqTopic topic: EnumMsgMqTopic.values()) { | |||||
| DirectExchange exchange = new DirectExchange(topic.getCode()); | |||||
| Queue queue = new Queue(topic.getCode()); | |||||
| Binding binding = BindingBuilder.bind(queue).to(exchange).withQueueName(); | |||||
| amqpAdmin.declareExchange(exchange); | |||||
| amqpAdmin.declareQueue(queue); | |||||
| amqpAdmin.declareBinding(binding); | |||||
| queuenames = queuenames +","+topic.getCode(); | |||||
| } | |||||
| if (queuenames.length()>0) { | |||||
| queuenames = queuenames.substring(0, queuenames.length()-1); | |||||
| } | |||||
| } | } | ||||
| @Autowired | @Autowired | ||||
| private AmqpTemplate rabbitTemplate; | private AmqpTemplate rabbitTemplate; | ||||
| public void send(String msg) { | |||||
| public void send(String queueName, String msg) { | |||||
| rabbitTemplate.convertAndSend(queueName, msg); | rabbitTemplate.convertAndSend(queueName, msg); | ||||
| } | } | ||||
| @RabbitListener(queues = "topic-1") | |||||
| @Bean | |||||
| public Queue fmRabbitmaQueue(){ | |||||
| return new Queue(queuenames,true); | |||||
| } | |||||
| @RabbitListener(queues = "#{fmRabbitmaQueue.name}") | |||||
| public void onMessage(String message) { | public void onMessage(String message) { | ||||
| doMessage(message); | doMessage(message); | ||||
| } | } | ||||