diff --git a/mallinkAdmin/src/main/java/com/iformall/UserApplication.java b/mallinkAdmin/src/main/java/com/iformall/UserApplication.java index f1c776d8b..f72f27524 100644 --- a/mallinkAdmin/src/main/java/com/iformall/UserApplication.java +++ b/mallinkAdmin/src/main/java/com/iformall/UserApplication.java @@ -1,6 +1,7 @@ package com.iformall; import com.ulisesbocchio.jasyptspringboot.annotation.EnableEncryptableProperties; +import org.rocketmq.starter.annotation.EnableRocketMQ; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -18,6 +19,7 @@ import tk.mybatis.spring.annotation.MapperScan; @EnableSwagger2 @EnableEncryptableProperties @EnableAsync +@EnableRocketMQ public class UserApplication { @Value("${fm.exception}") diff --git a/mallinkAdmin/src/main/resources/application.yml b/mallinkAdmin/src/main/resources/application.yml index b7a2d6448..147e79a5b 100644 --- a/mallinkAdmin/src/main/resources/application.yml +++ b/mallinkAdmin/src/main/resources/application.yml @@ -21,6 +21,15 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) + rocketmq: + nameServer: 127.0.0.1:9876 + producer: + retry-times-when-send-async-failed: 0 + send-msg-timeout: 300000 + compress-msg-body-over-howmuch: 4096 + max-message-size: 4194304 + retry-another-broker-when-not-store-ok: false + retry-times-when-send-failed: 2 # @{link} https://github.com/abel533 #Mybatis diff --git a/mallinkBApi/src/main/java/com/iformall/BApplication.java b/mallinkBApi/src/main/java/com/iformall/BApplication.java index ef3c776ba..0dabd6ccc 100644 --- a/mallinkBApi/src/main/java/com/iformall/BApplication.java +++ b/mallinkBApi/src/main/java/com/iformall/BApplication.java @@ -1,6 +1,7 @@ package com.iformall; import com.ulisesbocchio.jasyptspringboot.annotation.EnableEncryptableProperties; +import org.rocketmq.starter.annotation.EnableRocketMQ; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -14,6 +15,7 @@ import tk.mybatis.spring.annotation.MapperScan; @SpringBootApplication @MapperScan(basePackages = {"com.iformall.mapper"}) @EnableEncryptableProperties +@EnableRocketMQ public class BApplication { @Value("${fm.exception}") diff --git a/mallinkBApi/src/main/resources/application.yml b/mallinkBApi/src/main/resources/application.yml index 0e4588f52..c54a63361 100644 --- a/mallinkBApi/src/main/resources/application.yml +++ b/mallinkBApi/src/main/resources/application.yml @@ -12,6 +12,15 @@ spring: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null + rocketmq: + nameServer: 127.0.0.1:9876 + producer: + retry-times-when-send-async-failed: 0 + send-msg-timeout: 300000 + compress-msg-body-over-howmuch: 4096 + max-message-size: 4194304 + retry-another-broker-when-not-store-ok: false + retry-times-when-send-failed: 2 # @{link} https://github.com/abel533 #Mybatis diff --git a/mallinkCApi/src/main/java/com/iformall/CApplication.java b/mallinkCApi/src/main/java/com/iformall/CApplication.java index eca6c85b2..d2717e841 100644 --- a/mallinkCApi/src/main/java/com/iformall/CApplication.java +++ b/mallinkCApi/src/main/java/com/iformall/CApplication.java @@ -1,6 +1,7 @@ package com.iformall; import com.ulisesbocchio.jasyptspringboot.annotation.EnableEncryptableProperties; +import org.rocketmq.starter.annotation.EnableRocketMQ; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -14,6 +15,7 @@ import tk.mybatis.spring.annotation.MapperScan; @SpringBootApplication @MapperScan(basePackages = {"com.iformall.mapper"}) @EnableEncryptableProperties +@EnableRocketMQ public class CApplication { @Value("${fm.exception}") diff --git a/mallinkCApi/src/main/resources/application.yml b/mallinkCApi/src/main/resources/application.yml index bf31b2046..93efc61c0 100644 --- a/mallinkCApi/src/main/resources/application.yml +++ b/mallinkCApi/src/main/resources/application.yml @@ -12,6 +12,15 @@ spring: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null + rocketmq: + nameServer: 127.0.0.1:9876 + producer: + retry-times-when-send-async-failed: 0 + send-msg-timeout: 300000 + compress-msg-body-over-howmuch: 4096 + max-message-size: 4194304 + retry-another-broker-when-not-store-ok: false + retry-times-when-send-failed: 2 # @{link} https://github.com/abel533 #Mybatis diff --git a/mallinkSchedule/src/main/java/com/iformall/ScheduleApplication.java b/mallinkSchedule/src/main/java/com/iformall/ScheduleApplication.java index 68815c746..b9396d4e2 100644 --- a/mallinkSchedule/src/main/java/com/iformall/ScheduleApplication.java +++ b/mallinkSchedule/src/main/java/com/iformall/ScheduleApplication.java @@ -1,6 +1,7 @@ package com.iformall; import com.ulisesbocchio.jasyptspringboot.annotation.EnableEncryptableProperties; +import org.rocketmq.starter.annotation.EnableRocketMQ; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -15,6 +16,7 @@ import tk.mybatis.spring.annotation.MapperScan; @SpringBootApplication @MapperScan(basePackages = {"com.iformall.mapper"}) @EnableSwagger2 +@EnableRocketMQ @EnableEncryptableProperties public class ScheduleApplication { diff --git a/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java b/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java index 6ace453ad..2b260585c 100644 --- a/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java +++ b/mallinkSchedule/src/main/java/com/iformall/controller/HomeController.java @@ -1,10 +1,12 @@ package com.iformall.controller; import com.iformall.common.ResultData; +import com.iformall.mq.MessageProducer; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; 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.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; @@ -18,10 +20,19 @@ public class HomeController { @Value("${version}") private String version; + @Autowired + private MessageProducer messageProducer; + @ApiOperation("获取后端版本号") @GetMapping("/version") public ResultData version() { return new ResultData(version); } + + @GetMapping("/mqtest") + public ResultData mqtest() { + messageProducer.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.yml b/mallinkSchedule/src/main/resources/application.yml index 14e568b8c..3adc484c3 100644 --- a/mallinkSchedule/src/main/resources/application.yml +++ b/mallinkSchedule/src/main/resources/application.yml @@ -21,7 +21,15 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) - + rocketmq: + nameServer: 127.0.0.1:9876 + producer: + retry-times-when-send-async-failed: 0 + send-msg-timeout: 300000 + compress-msg-body-over-howmuch: 4096 + max-message-size: 4194304 + retry-another-broker-when-not-store-ok: false + retry-times-when-send-failed: 2 # @{link} https://github.com/abel533 diff --git a/mallinkService/src/main/java/com/iformall/mq/MessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/MessageProducer.java new file mode 100644 index 000000000..ea69a889b --- /dev/null +++ b/mallinkService/src/main/java/com/iformall/mq/MessageProducer.java @@ -0,0 +1,60 @@ +package com.iformall.mq; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import javax.annotation.PostConstruct; + +/** + * @Auther: Stormeye + * @Date: 2019-01-31 + * @Description: RocketMQ消息生产者 + */ +@Service +@Slf4j +public class MessageProducer { + @Value("${spring.rocketmq.nameServer}") + private String namesrvAddr; + + private static final DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer"); + + @PostConstruct + public void start(){ + try { + producer.setNamesrvAddr(namesrvAddr); + producer.start(); + log.info("Message Producer Start..."); + }catch (Exception e){ + log.error("Message Producer Start Error!!",e); + } + } + + public void sendMessage(String data, String topic, String tags, String keys) { + try { + byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); + + Message mqMsg = new Message(topic, tags, keys, messageBody); + + producer.send(mqMsg, new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + log.info("Message Producer: Send Message Success {}", sendResult); + } + + @Override + public void onException(Throwable throwable) { + log.error("Message Producer: Send Message Error ", throwable); + } + }); + } catch (Exception e) { + log.error("Message Producer: Send Message Error ", e); + } + + } +} diff --git a/mallinkService/src/main/java/com/iformall/mq/MqConsumer.java b/mallinkService/src/main/java/com/iformall/mq/MqConsumer.java new file mode 100644 index 000000000..8c9a7789c --- /dev/null +++ b/mallinkService/src/main/java/com/iformall/mq/MqConsumer.java @@ -0,0 +1,17 @@ +package com.iformall.mq; + +import lombok.extern.slf4j.Slf4j; +import org.rocketmq.starter.annotation.RocketMQListener; +import org.rocketmq.starter.annotation.RocketMQMessage; +import org.springframework.stereotype.Service; + +@Slf4j +@Service +@RocketMQListener(topic = "topic-1") +public class MqConsumer { + + @RocketMQMessage(messageClass = String.class, tag = "tag-1") + public void onMessage(String message) { + log.info("received message: {}", message); + } +} diff --git a/pom.xml b/pom.xml index a3c7f2e90..0bd9efe42 100644 --- a/pom.xml +++ b/pom.xml @@ -118,6 +118,18 @@ test + + + org.rocketmq.spring.boot + rocketmq-spring-boot-starter + 1.0.0.RELEASE + + + org.apache.rocketmq + rocketmq-client + 4.4.0 + + org.apache.commons commons-lang3