From 1964bcb4beddd1fd31c59ea8eae3c63ac1996c99 Mon Sep 17 00:00:00 2001 From: Stormeye Wu Date: Tue, 5 Mar 2019 19:51:43 +0800 Subject: [PATCH] =?UTF-8?q?[MQ][=E4=BF=AE=E6=94=B9]:dev=E5=90=AF=E7=94=A8r?= =?UTF-8?q?abbitmq=EF=BC=8Ctest/prod=E5=90=AF=E7=94=A8rocketmq?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-dev.yml | 2 ++ .../src/main/resources/application-prod.yml | 2 ++ .../src/main/resources/application-test.yml | 2 ++ .../src/main/resources/application.yml | 7 +++++ .../iformall/controller/HomeController.java | 11 +++++-- .../src/main/resources/application-dev.yml | 2 ++ .../src/main/resources/application.yml | 7 +++++ .../main/java/com/iformall/mq/MQConfig.java | 4 +-- .../java/com/iformall/mq/MqBaseProducer.java | 6 ++++ .../iformall/mq/impl/RabbitMqConsumer.java | 29 ++++++++++++++--- .../mq/impl/RabbitMqMessageProducer.java | 31 +++++++++++++++++++ .../mq/impl/RocketMqMessageProducer.java | 9 ++++++ 12 files changed, 104 insertions(+), 8 deletions(-) create mode 100644 mallinkService/src/main/java/com/iformall/mq/MqBaseProducer.java create mode 100644 mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java diff --git a/mallinkAdmin/src/main/resources/application-dev.yml b/mallinkAdmin/src/main/resources/application-dev.yml index 5c74d058d..4e12234d3 100644 --- a/mallinkAdmin/src/main/resources/application-dev.yml +++ b/mallinkAdmin/src/main/resources/application-dev.yml @@ -1,4 +1,6 @@ spring: + profiles: + include: rabbitMQ # JDBC datasource: url: jdbc:mysql://202.165.179.86:3306/mallinkDev?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false diff --git a/mallinkAdmin/src/main/resources/application-prod.yml b/mallinkAdmin/src/main/resources/application-prod.yml index a7ff32371..4ae7eb723 100644 --- a/mallinkAdmin/src/main/resources/application-prod.yml +++ b/mallinkAdmin/src/main/resources/application-prod.yml @@ -1,4 +1,6 @@ spring: + profiles: + include: rocketMQ # JDBC datasource: url: jdbc:mysql://formalldb.cfqyqflkdlit.rds.cn-northwest-1.amazonaws.com.cn:3306/mallink?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&useSSL=false diff --git a/mallinkAdmin/src/main/resources/application-test.yml b/mallinkAdmin/src/main/resources/application-test.yml index a04d8abe4..fe823f7fe 100644 --- a/mallinkAdmin/src/main/resources/application-test.yml +++ b/mallinkAdmin/src/main/resources/application-test.yml @@ -1,4 +1,6 @@ spring: + profiles: + include: rocketMQ # JDBC datasource: url: jdbc:mysql://formalldb.cfqyqflkdlit.rds.cn-northwest-1.amazonaws.com.cn:3306/mallinkTest?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&useSSL=false diff --git a/mallinkAdmin/src/main/resources/application.yml b/mallinkAdmin/src/main/resources/application.yml index a9fea51f1..7f5ab94eb 100644 --- a/mallinkAdmin/src/main/resources/application.yml +++ b/mallinkAdmin/src/main/resources/application.yml @@ -30,6 +30,13 @@ spring: max-message-size: 4194304 retry-another-broker-when-not-store-ok: false retry-times-when-send-failed: 2 + rabbitmq: + host: 127.0.0.1 + port: 5672 + username: guest + password: guest + publisher-confirms: true + virtual-host: / # @{link} https://github.com/abel533 diff --git a/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java b/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java index a927d10bb..6e106265f 100644 --- a/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java +++ b/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java @@ -1,6 +1,9 @@ package com.iformall.controller; import com.iformall.common.ResultData; +import com.iformall.mq.MQConfig; +import com.iformall.mq.MqBaseProducer; +import com.iformall.mq.impl.RabbitMqMessageProducer; import com.iformall.mq.impl.RocketMqMessageProducer; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; @@ -8,6 +11,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; @@ -21,7 +25,10 @@ public class HomeController { private String version; @Autowired - private RocketMqMessageProducer messageProducer; + private MqBaseProducer mqBaseProducer; + + @Autowired + private RabbitMqMessageProducer rabbitMqMessageProducer; @ApiOperation("获取后端版本号") @@ -32,7 +39,7 @@ public class HomeController { @GetMapping("/mqtest") public ResultData mqtest() { - messageProducer.sendMessage("mq test 1", "topic-1", "tag-1", "key-1"); + mqBaseProducer.sendMessage("mq test 1", "topic-1", "tag-1", "key-1"); return new ResultData(); } } \ No newline at end of file diff --git a/mallinkSchedule/src/main/resources/application-dev.yml b/mallinkSchedule/src/main/resources/application-dev.yml index 70c47007f..a7c1dadcf 100644 --- a/mallinkSchedule/src/main/resources/application-dev.yml +++ b/mallinkSchedule/src/main/resources/application-dev.yml @@ -1,4 +1,6 @@ spring: + profiles: + include: rabbitMQ # JDBC datasource: url: jdbc:mysql://202.165.179.86:3306/mallinkDev?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false diff --git a/mallinkSchedule/src/main/resources/application.yml b/mallinkSchedule/src/main/resources/application.yml index 3adc484c3..482a50e6d 100644 --- a/mallinkSchedule/src/main/resources/application.yml +++ b/mallinkSchedule/src/main/resources/application.yml @@ -30,6 +30,13 @@ spring: max-message-size: 4194304 retry-another-broker-when-not-store-ok: false retry-times-when-send-failed: 2 + rabbitmq: + host: 127.0.0.1 + port: 5672 + username: guest + password: guest + publisher-confirms: true + virtual-host: / # @{link} https://github.com/abel533 diff --git a/mallinkService/src/main/java/com/iformall/mq/MQConfig.java b/mallinkService/src/main/java/com/iformall/mq/MQConfig.java index 4eda45290..ffdc5daf1 100644 --- a/mallinkService/src/main/java/com/iformall/mq/MQConfig.java +++ b/mallinkService/src/main/java/com/iformall/mq/MQConfig.java @@ -2,7 +2,7 @@ package com.iformall.mq; public class MQConfig { public static class Impl { - public static final String RABBIT_MQ = "rabbitmq"; - public static final String ROCKET_MQ = "rocketmq"; + public static final String RABBIT_MQ = "rabbitMQ"; + public static final String ROCKET_MQ = "rocketMQ"; } } diff --git a/mallinkService/src/main/java/com/iformall/mq/MqBaseProducer.java b/mallinkService/src/main/java/com/iformall/mq/MqBaseProducer.java new file mode 100644 index 000000000..476e0974b --- /dev/null +++ b/mallinkService/src/main/java/com/iformall/mq/MqBaseProducer.java @@ -0,0 +1,6 @@ +package com.iformall.mq; + +public interface MqBaseProducer { + + void sendMessage(String data, String topic, String tags, String keys); +} diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqConsumer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqConsumer.java index d912daafd..bd86d2826 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqConsumer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqConsumer.java @@ -2,17 +2,38 @@ package com.iformall.mq.impl; import com.iformall.mq.MQConfig; import com.iformall.mq.MqBaseConsumer; -import lombok.extern.slf4j.Slf4j; -import org.rocketmq.starter.annotation.RocketMQListener; -import org.rocketmq.starter.annotation.RocketMQMessage; +import org.springframework.amqp.core.*; import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Component; -import org.springframework.stereotype.Service; + +import javax.annotation.PostConstruct; @Component @Profile(MQConfig.Impl.RABBIT_MQ) public class RabbitMqConsumer extends MqBaseConsumer { + String queueName = "topic-1"; + + @Autowired + private AmqpAdmin amqpAdmin; + + @PostConstruct + 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); + } + + @Autowired + private AmqpTemplate rabbitTemplate; + + public void send(String msg) { + rabbitTemplate.convertAndSend(queueName, msg); + } @RabbitListener(queues = "topic-1") public void onMessage(String message) { diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java new file mode 100644 index 000000000..9332ba44e --- /dev/null +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java @@ -0,0 +1,31 @@ +package com.iformall.mq.impl; + +import com.iformall.mq.MQConfig; +import com.iformall.mq.MqBaseProducer; +import lombok.extern.slf4j.Slf4j; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; + +/** + * @Auther: Stormeye + * @Date: 2019-01-31 + * @Description: RocketMQ消息生产者 + */ +@Service +@Slf4j +@Profile(MQConfig.Impl.RABBIT_MQ) +public class RabbitMqMessageProducer implements MqBaseProducer { + @Autowired + private AmqpTemplate rabbitTemplate; + + @Override + public void sendMessage(String data, String topic, String tags, String keys) { + try { + rabbitTemplate.convertAndSend(topic, data); + } catch (Exception e) { + log.error("RabbitMqMessageProducer: Send Message Error ", e); + } + } +} diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java index 993c1b9a5..cf1a54df6 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java @@ -1,5 +1,7 @@ package com.iformall.mq.impl; +import com.iformall.mq.MQConfig; +import com.iformall.mq.MqBaseProducer; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendCallback; @@ -9,6 +11,7 @@ import org.apache.rocketmq.remoting.common.RemotingHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; @@ -20,6 +23,7 @@ import javax.annotation.PostConstruct; */ @Service @Slf4j +<<<<<<< HEAD <<<<<<< HEAD:mallinkService/src/main/java/com/iformall/mq/MessageProducer.java public class MessageProducer { private final Logger log = LoggerFactory.getLogger(this.getClass()); @@ -27,6 +31,10 @@ public class MessageProducer { ======= public class RocketMqMessageProducer { >>>>>>> add rabbitmq:mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java +======= +@Profile(MQConfig.Impl.ROCKET_MQ) +public class RocketMqMessageProducer implements MqBaseProducer { +>>>>>>> [MQ][修改]:dev启用rabbitmq,test/prod启用rocketmq @Value("${spring.rocketmq.nameServer}") private String namesrvAddr; @@ -43,6 +51,7 @@ public class RocketMqMessageProducer { } } + @Override public void sendMessage(String data, String topic, String tags, String keys) { try { byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);