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);