| @@ -1,4 +1,6 @@ | |||||
| spring: | spring: | ||||
| profiles: | |||||
| include: rabbitMQ | |||||
| # JDBC | # JDBC | ||||
| datasource: | datasource: | ||||
| url: jdbc:mysql://202.165.179.86:3306/mallinkDev?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false | url: jdbc:mysql://202.165.179.86:3306/mallinkDev?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false | ||||
| @@ -1,4 +1,6 @@ | |||||
| spring: | spring: | ||||
| profiles: | |||||
| include: rocketMQ | |||||
| # JDBC | # JDBC | ||||
| datasource: | 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 | url: jdbc:mysql://formalldb.cfqyqflkdlit.rds.cn-northwest-1.amazonaws.com.cn:3306/mallink?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&useSSL=false | ||||
| @@ -1,4 +1,6 @@ | |||||
| spring: | spring: | ||||
| profiles: | |||||
| include: rocketMQ | |||||
| # JDBC | # JDBC | ||||
| datasource: | 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 | url: jdbc:mysql://formalldb.cfqyqflkdlit.rds.cn-northwest-1.amazonaws.com.cn:3306/mallinkTest?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&useSSL=false | ||||
| @@ -30,6 +30,13 @@ spring: | |||||
| max-message-size: 4194304 | max-message-size: 4194304 | ||||
| retry-another-broker-when-not-store-ok: false | retry-another-broker-when-not-store-ok: false | ||||
| retry-times-when-send-failed: 2 | 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 | # @{link} https://github.com/abel533 | ||||
| @@ -1,6 +1,9 @@ | |||||
| package com.iformall.controller; | package com.iformall.controller; | ||||
| import com.iformall.common.ResultData; | 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 com.iformall.mq.impl.RocketMqMessageProducer; | ||||
| import io.swagger.annotations.Api; | import io.swagger.annotations.Api; | ||||
| import io.swagger.annotations.ApiOperation; | import io.swagger.annotations.ApiOperation; | ||||
| @@ -8,6 +11,7 @@ import org.slf4j.Logger; | |||||
| import org.slf4j.LoggerFactory; | import org.slf4j.LoggerFactory; | ||||
| import org.springframework.beans.factory.annotation.Autowired; | import org.springframework.beans.factory.annotation.Autowired; | ||||
| import org.springframework.beans.factory.annotation.Value; | 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.GetMapping; | ||||
| import org.springframework.web.bind.annotation.RestController; | import org.springframework.web.bind.annotation.RestController; | ||||
| @@ -21,7 +25,10 @@ public class HomeController { | |||||
| private String version; | private String version; | ||||
| @Autowired | @Autowired | ||||
| private RocketMqMessageProducer messageProducer; | |||||
| private MqBaseProducer mqBaseProducer; | |||||
| @Autowired | |||||
| private RabbitMqMessageProducer rabbitMqMessageProducer; | |||||
| @ApiOperation("获取后端版本号") | @ApiOperation("获取后端版本号") | ||||
| @@ -32,7 +39,7 @@ public class HomeController { | |||||
| @GetMapping("/mqtest") | @GetMapping("/mqtest") | ||||
| public ResultData 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(); | return new ResultData(); | ||||
| } | } | ||||
| } | } | ||||
| @@ -1,4 +1,6 @@ | |||||
| spring: | spring: | ||||
| profiles: | |||||
| include: rabbitMQ | |||||
| # JDBC | # JDBC | ||||
| datasource: | datasource: | ||||
| url: jdbc:mysql://202.165.179.86:3306/mallinkDev?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false | url: jdbc:mysql://202.165.179.86:3306/mallinkDev?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false | ||||
| @@ -30,6 +30,13 @@ spring: | |||||
| max-message-size: 4194304 | max-message-size: 4194304 | ||||
| retry-another-broker-when-not-store-ok: false | retry-another-broker-when-not-store-ok: false | ||||
| retry-times-when-send-failed: 2 | 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 | # @{link} https://github.com/abel533 | ||||
| @@ -2,7 +2,7 @@ package com.iformall.mq; | |||||
| public class MQConfig { | public class MQConfig { | ||||
| public static class Impl { | 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"; | |||||
| } | } | ||||
| } | } | ||||
| @@ -0,0 +1,6 @@ | |||||
| package com.iformall.mq; | |||||
| public interface MqBaseProducer { | |||||
| void sendMessage(String data, String topic, String tags, String keys); | |||||
| } | |||||
| @@ -2,17 +2,38 @@ package com.iformall.mq.impl; | |||||
| import com.iformall.mq.MQConfig; | import com.iformall.mq.MQConfig; | ||||
| import com.iformall.mq.MqBaseConsumer; | 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.amqp.rabbit.annotation.RabbitListener; | ||||
| import org.springframework.beans.factory.annotation.Autowired; | |||||
| import org.springframework.context.annotation.Profile; | import org.springframework.context.annotation.Profile; | ||||
| import org.springframework.stereotype.Component; | import org.springframework.stereotype.Component; | ||||
| import org.springframework.stereotype.Service; | |||||
| 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"; | |||||
| @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") | @RabbitListener(queues = "topic-1") | ||||
| public void onMessage(String message) { | public void onMessage(String message) { | ||||
| @@ -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); | |||||
| } | |||||
| } | |||||
| } | |||||
| @@ -1,5 +1,7 @@ | |||||
| package com.iformall.mq.impl; | package com.iformall.mq.impl; | ||||
| import com.iformall.mq.MQConfig; | |||||
| import com.iformall.mq.MqBaseProducer; | |||||
| import lombok.extern.slf4j.Slf4j; | import lombok.extern.slf4j.Slf4j; | ||||
| import org.apache.rocketmq.client.producer.DefaultMQProducer; | import org.apache.rocketmq.client.producer.DefaultMQProducer; | ||||
| import org.apache.rocketmq.client.producer.SendCallback; | 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.Logger; | ||||
| import org.slf4j.LoggerFactory; | import org.slf4j.LoggerFactory; | ||||
| import org.springframework.beans.factory.annotation.Value; | import org.springframework.beans.factory.annotation.Value; | ||||
| import org.springframework.context.annotation.Profile; | |||||
| import org.springframework.stereotype.Service; | import org.springframework.stereotype.Service; | ||||
| import javax.annotation.PostConstruct; | import javax.annotation.PostConstruct; | ||||
| @@ -20,6 +23,7 @@ import javax.annotation.PostConstruct; | |||||
| */ | */ | ||||
| @Service | @Service | ||||
| @Slf4j | @Slf4j | ||||
| <<<<<<< HEAD | |||||
| <<<<<<< HEAD:mallinkService/src/main/java/com/iformall/mq/MessageProducer.java | <<<<<<< HEAD:mallinkService/src/main/java/com/iformall/mq/MessageProducer.java | ||||
| public class MessageProducer { | public class MessageProducer { | ||||
| private final Logger log = LoggerFactory.getLogger(this.getClass()); | private final Logger log = LoggerFactory.getLogger(this.getClass()); | ||||
| @@ -27,6 +31,10 @@ public class MessageProducer { | |||||
| ======= | ======= | ||||
| public class RocketMqMessageProducer { | public class RocketMqMessageProducer { | ||||
| >>>>>>> add rabbitmq:mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java | >>>>>>> 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}") | @Value("${spring.rocketmq.nameServer}") | ||||
| private String namesrvAddr; | private String namesrvAddr; | ||||
| @@ -43,6 +51,7 @@ public class RocketMqMessageProducer { | |||||
| } | } | ||||
| } | } | ||||
| @Override | |||||
| public void sendMessage(String data, String topic, String tags, String keys) { | public void sendMessage(String data, String topic, String tags, String keys) { | ||||
| try { | try { | ||||
| byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); | byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); | ||||