From 068c5b7cbf239d6aae87fe8ae6239f1165f76431 Mon Sep 17 00:00:00 2001
From: Administrator <admin>
Date: Tue, 12 Apr 2022 19:13:21 +0800
Subject: [PATCH] 新增 rabbitmq 设置
---
src/main/java/org/springblade/modules/rabbitmq/controller/SendMessageController.java | 81 ++++++++
src/main/java/org/springblade/modules/rabbitmq/config/KeyPrefix.java | 8
src/main/java/org/springblade/modules/rabbitmq/redis/BasePrefix.java | 31 +++
src/main/java/org/springblade/modules/rabbitmq/vo/TaskMessage.java | 13 +
src/main/java/org/springblade/modules/rabbitmq/config/MQConfig.java | 64 ++++++
src/main/java/org/springblade/modules/rabbitmq/redis/RedisConfig.java | 68 ++++++
src/main/resources/application-test.yml | 21 +
pom.xml | 4
src/main/java/org/springblade/modules/rabbitmq/redis/RedisPoolFactory.java | 26 ++
Dockerfile | 2
src/main/java/org/springblade/modules/rabbitmq/redis/TaskKey.java | 10 +
src/main/java/org/springblade/modules/rabbitmq/redis/RedisService.java | 141 ++++++++++++++
src/main/java/org/springblade/modules/rabbitmq/config/MQreceiver.java | 34 +++
src/main/java/org/springblade/modules/rabbitmq/vo/Task.java | 21 ++
src/main/java/org/springblade/modules/rabbitmq/config/MQsender.java | 37 +++
src/main/java/org/springblade/modules/rabbitmq/vo/Man.java | 16 +
16 files changed, 573 insertions(+), 4 deletions(-)
diff --git a/Dockerfile b/Dockerfile
index ac8e5ed..3d40288 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -3,7 +3,7 @@
MAINTAINER smallchill@163.com
#启动自行加载 服务名-prod.yml配置
-ENV PARAMS="--server.port=81 --spring.profiles.active=test"
+ENV PARAMS="--server.port=81 --spring.profiles.active=prod"
RUN /bin/cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && echo 'Asia/Shanghai' >/etc/timezone
COPY target/zhbaw.jar /app.jar
diff --git a/pom.xml b/pom.xml
index 0f52159..caf2bcc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -290,6 +290,10 @@
<!-- <artifactId>ureport-spring-boot-starter</artifactId>-->
<!-- <version>2.2.9</version>-->
<!-- </dependency>-->
+ <dependency>
+ <groupId>redis.clients</groupId>
+ <artifactId>jedis</artifactId>
+ </dependency>
</dependencies>
<profiles>
diff --git a/src/main/java/org/springblade/modules/rabbitmq/config/KeyPrefix.java b/src/main/java/org/springblade/modules/rabbitmq/config/KeyPrefix.java
new file mode 100644
index 0000000..cb6e081
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/config/KeyPrefix.java
@@ -0,0 +1,8 @@
+package org.springblade.modules.rabbitmq.config;
+
+public interface KeyPrefix {
+
+ public int expireSeconds();
+
+ public String getPrefix();
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/config/MQConfig.java b/src/main/java/org/springblade/modules/rabbitmq/config/MQConfig.java
new file mode 100644
index 0000000..789d961
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/config/MQConfig.java
@@ -0,0 +1,64 @@
+package org.springblade.modules.rabbitmq.config;
+
+import org.springframework.amqp.core.*;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * rabbitmq 配置文件
+ * @author zhongrj
+ * @since 2022-04-08
+ */
+@Configuration
+public class MQConfig {
+ public static final String QUEUE = "queue";
+ public static final String TOPIC_QUEUE1= "topic.queue1";
+ public static final String TOPIC_QUEUE2 = "topic.queue2";
+ public static final String TOPIC_EXCHANGE = "topicExchange";
+ public static final String FANOUT_EXCHANGE = "fanoutxchage";
+ public static final String TASK_QUEUE = "task_queue";
+
+ @Bean
+ public Queue taskQueue(){
+ return new Queue(TASK_QUEUE,true);
+ }
+
+ //Direct模式
+ @Bean
+ public Queue queue(){
+ return new Queue(QUEUE,true);
+ }
+
+ //topic交换机模式
+ @Bean
+ public Queue topicQueue1(){
+ return new Queue(TOPIC_QUEUE1,true);
+ }
+
+ @Bean
+ public Queue topicQueue2(){
+ return new Queue(TOPIC_QUEUE2,true);
+ }
+
+ @Bean
+ public TopicExchange topicExchange(){
+ return new TopicExchange(TOPIC_EXCHANGE);
+ }
+
+ @Bean
+ public Binding topicBinding1(){
+ return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("topic.key1");
+ }
+
+ @Bean
+ public Binding topicBinding2(){
+ return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.#");
+ }
+
+ //Fanout模式，广播模式
+
+ @Bean
+ public FanoutExchange fanoutExchange(){
+ return new FanoutExchange(FANOUT_EXCHANGE);
+ }
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/config/MQreceiver.java b/src/main/java/org/springblade/modules/rabbitmq/config/MQreceiver.java
new file mode 100644
index 0000000..c1a07d0
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/config/MQreceiver.java
@@ -0,0 +1,34 @@
+package org.springblade.modules.rabbitmq.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springblade.modules.rabbitmq.redis.RedisService;
+import org.springblade.modules.rabbitmq.vo.TaskMessage;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.stereotype.Service;
+
+/**
+ * 队列消息接收
+ * @author zhongrj
+ * @since 2022-04-12
+ */
+@Service
+public class MQreceiver {
+
+
+ private static Logger log = LoggerFactory.getLogger(MQreceiver.class);
+
+
+ @RabbitListener(queues = MQConfig.TASK_QUEUE)
+ public void receive(String message){
+
+ log.info("receive message:"+message);
+
+
+ TaskMessage taskMessage = RedisService.stringToBean(message, TaskMessage.class);
+
+ System.out.println("用户" + taskMessage.getManId() + "抢到了任务："+ taskMessage.getTaskId());
+
+ }
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/config/MQsender.java b/src/main/java/org/springblade/modules/rabbitmq/config/MQsender.java
new file mode 100644
index 0000000..089eaca
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/config/MQsender.java
@@ -0,0 +1,37 @@
+package org.springblade.modules.rabbitmq.config;
+
+import org.springblade.modules.rabbitmq.redis.RedisService;
+import org.springblade.modules.rabbitmq.vo.TaskMessage;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * 消息发送
+ * @author zhongrj
+ * @since 2022-04-12
+ *
+ */
+@Service
+public class MQsender {
+
+ @Autowired
+ AmqpTemplate amqpTemplate;
+
+
+ public void send(Object message){
+ String msg = RedisService.beanToString(message);
+ amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);
+ }
+
+ public void sendTaskMessage(TaskMessage message){
+ String msg = RedisService.beanToString(message);
+ amqpTemplate.convertAndSend(MQConfig.TASK_QUEUE,msg);
+ }
+
+ public void sendTopic(Object message){
+ String msg = RedisService.beanToString(message);
+ amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key1",msg+"1");
+ }
+
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/controller/SendMessageController.java b/src/main/java/org/springblade/modules/rabbitmq/controller/SendMessageController.java
new file mode 100644
index 0000000..bcebf74
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/controller/SendMessageController.java
@@ -0,0 +1,81 @@
+package org.springblade.modules.rabbitmq.controller;
+
+import lombok.AllArgsConstructor;
+import org.springblade.modules.rabbitmq.config.MQsender;
+import org.springblade.modules.rabbitmq.redis.TaskKey;
+import org.springblade.modules.rabbitmq.redis.RedisService;
+import org.springblade.modules.rabbitmq.vo.Man;
+import org.springblade.modules.rabbitmq.vo.Task;
+import org.springblade.modules.rabbitmq.vo.TaskMessage;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Random;
+
+/**
+ * 消息发送
+ * @author zhongrj
+ */
+@RestController
+@AllArgsConstructor
+@RequestMapping("/rabbitmq")
+public class SendMessageController {
+
+ private static int count = 0;
+
+ private final MQsender mQsender;
+
+ private final RedisService redisService;
+
+
+ @GetMapping("/sendDirectMessage")
+ public String sendDirectMessage() {
+ String messageData = "test message, hello!";
+ mQsender.send(messageData);
+ return "ok";
+ }
+
+ /**
+ * 模拟抢单
+ * @return
+ */
+ @GetMapping("/createTask")
+ public String createTask(Long id ,Integer num) {
+ //创建任务
+ Task task = new Task();
+ task.setId(id);
+ task.setNum(num);
+ task.setTaskName("测试");
+ //将任务数量和对应的id 存入redis
+ redisService.set(TaskKey.getTaskStock,""+task.getId(),task.getNum());
+ //返回
+ return "ok";
+ }
+
+
+ /**
+ * 模拟抢单（不同人员同时抢）
+ * @return
+ */
+ @GetMapping("/testSend")
+ public String testSend(Long id) {
+ //模拟创建人员
+ Man man = new Man();
+ man.setId(System.currentTimeMillis() + new Random().nextInt(1000000));
+ man.setUsername("task"+ count);
+ // 预减库存
+ long stack = redisService.decr(TaskKey.getTaskStock,""+id);
+ if (stack<0){
+ return "已结束";
+ }
+ //入队列
+ TaskMessage taskMessage = new TaskMessage();
+ taskMessage.setTaskId(id);
+ taskMessage.setManId(man.getId());
+ //发送消息
+ mQsender.sendTaskMessage(taskMessage);
+ //返回
+ return "ok";
+ }
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/redis/BasePrefix.java b/src/main/java/org/springblade/modules/rabbitmq/redis/BasePrefix.java
new file mode 100644
index 0000000..00148f4
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/redis/BasePrefix.java
@@ -0,0 +1,31 @@
+package org.springblade.modules.rabbitmq.redis;
+
+import org.springblade.modules.rabbitmq.config.KeyPrefix;
+
+public abstract class BasePrefix implements KeyPrefix {
+
+ private int expireSeconds;
+
+ private String prefix;
+
+ public BasePrefix(String prefix) {//0代表永不过期
+ this(0, prefix);
+ }
+
+ public BasePrefix( int expireSeconds, String prefix) {
+ this.expireSeconds = expireSeconds;
+ this.prefix = prefix;
+ }
+
+ @Override
+ public int expireSeconds() {//默认0代表永不过期
+ return expireSeconds;
+ }
+
+ @Override
+ public String getPrefix() {
+ String className = getClass().getSimpleName();
+ return className+":" + prefix;
+ }
+
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/redis/RedisConfig.java b/src/main/java/org/springblade/modules/rabbitmq/redis/RedisConfig.java
new file mode 100644
index 0000000..ddaba2d
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/redis/RedisConfig.java
@@ -0,0 +1,68 @@
+package org.springblade.modules.rabbitmq.redis;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+@Component
+@ConfigurationProperties(prefix="redis")
+public class RedisConfig {
+
+ private String host;
+
+ private int port;
+ /**
+ * 秒
+ */
+ private int timeout;
+
+ private String password;
+
+ private int poolMaxTotal;
+
+ private int poolMaxIdle;
+ //秒
+ private int poolMaxWait;
+
+ public String getHost() {
+ return host;
+ }
+ public void setHost(String host) {
+ this.host = host;
+ }
+ public int getPort() {
+ return port;
+ }
+ public void setPort(int port) {
+ this.port = port;
+ }
+ public int getTimeout() {
+ return timeout;
+ }
+ public void setTimeout(int timeout) {
+ this.timeout = timeout;
+ }
+ public String getPassword() {
+ return password;
+ }
+ public void setPassword(String password) {
+ this.password = password;
+ }
+ public int getPoolMaxTotal() {
+ return poolMaxTotal;
+ }
+ public void setPoolMaxTotal(int poolMaxTotal) {
+ this.poolMaxTotal = poolMaxTotal;
+ }
+ public int getPoolMaxIdle() {
+ return poolMaxIdle;
+ }
+ public void setPoolMaxIdle(int poolMaxIdle) {
+ this.poolMaxIdle = poolMaxIdle;
+ }
+ public int getPoolMaxWait() {
+ return poolMaxWait;
+ }
+ public void setPoolMaxWait(int poolMaxWait) {
+ this.poolMaxWait = poolMaxWait;
+ }
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/redis/RedisPoolFactory.java b/src/main/java/org/springblade/modules/rabbitmq/redis/RedisPoolFactory.java
new file mode 100644
index 0000000..71bc78c
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/redis/RedisPoolFactory.java
@@ -0,0 +1,26 @@
+package org.springblade.modules.rabbitmq.redis;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Service;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+
+@Service
+public class RedisPoolFactory {
+
+ @Autowired
+ RedisConfig redisConfig;
+
+ @Bean
+ public JedisPool JedisPoolFactory() {
+ JedisPoolConfig poolConfig = new JedisPoolConfig();
+ poolConfig.setMaxIdle(redisConfig.getPoolMaxIdle());
+ poolConfig.setMaxTotal(redisConfig.getPoolMaxTotal());
+ poolConfig.setMaxWaitMillis(redisConfig.getPoolMaxWait() * 1000);
+ JedisPool jp = new JedisPool(poolConfig, redisConfig.getHost(), redisConfig.getPort(),
+ redisConfig.getTimeout()*1000);
+ return jp;
+ }
+
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/redis/RedisService.java b/src/main/java/org/springblade/modules/rabbitmq/redis/RedisService.java
new file mode 100644
index 0000000..c5197dd
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/redis/RedisService.java
@@ -0,0 +1,141 @@
+package org.springblade.modules.rabbitmq.redis;
+
+import com.alibaba.fastjson.JSON;
+import org.springblade.modules.rabbitmq.config.KeyPrefix;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+
+@Service
+public class RedisService {
+
+ @Autowired
+ private JedisPool jedisPool;
+
+ /**
+ * 获取当个对象
+ * */
+ public <T> T get(KeyPrefix prefix, String key, Class<T> clazz) {
+ Jedis jedis = null;
+ try {
+ jedis = jedisPool.getResource();
+ //生成真正的key
+ String realKey = prefix.getPrefix() + key;
+ String str = jedis.get(realKey);
+ T t = stringToBean(str, clazz);
+ return t;
+ }finally {
+ returnToPool(jedis);
+ }
+ }
+
+ /**
+ * 设置对象
+ * */
+ public <T> boolean set(KeyPrefix prefix, String key, T value) {
+ Jedis jedis = null;
+ try {
+ jedis = jedisPool.getResource();
+ String str = beanToString(value);
+ if(str == null || str.length() <= 0) {
+ return false;
+ }
+ //生成真正的key
+ String realKey = prefix.getPrefix() + key;
+ int seconds = prefix.expireSeconds();
+ if(seconds <= 0) {
+ jedis.set(realKey, str);
+ }else {
+ jedis.setex(realKey, seconds, str);
+ }
+ return true;
+ }finally {
+ returnToPool(jedis);
+ }
+ }
+
+ /**
+ * 判断key是否存在
+ * */
+ public <T> boolean exists(KeyPrefix prefix, String key) {
+ Jedis jedis = null;
+ try {
+ jedis = jedisPool.getResource();
+ //生成真正的key
+ String realKey = prefix.getPrefix() + key;
+ return jedis.exists(realKey);
+ }finally {
+ returnToPool(jedis);
+ }
+ }
+
+ /**
+ * 增加值
+ * */
+ public <T> Long incr(KeyPrefix prefix, String key) {
+ Jedis jedis = null;
+ try {
+ jedis = jedisPool.getResource();
+ //生成真正的key
+ String realKey = prefix.getPrefix() + key;
+ return jedis.incr(realKey);
+ }finally {
+ returnToPool(jedis);
+ }
+ }
+
+ /**
+ * 减少值
+ * */
+ public <T> Long decr(KeyPrefix prefix, String key) {
+ Jedis jedis = null;
+ try {
+ jedis = jedisPool.getResource();
+ //生成真正的key
+ String realKey = prefix.getPrefix() + key;
+ return jedis.decr(realKey);
+ }finally {
+ returnToPool(jedis);
+ }
+ }
+
+ public static <T> String beanToString(T value) {
+ if(value == null) {
+ return null;
+ }
+ Class<?> clazz = value.getClass();
+ if(clazz == int.class || clazz == Integer.class) {
+ return ""+value;
+ }else if(clazz == String.class) {
+ return (String)value;
+ }else if(clazz == long.class || clazz == Long.class) {
+ return ""+value;
+ }else {
+ return JSON.toJSONString(value);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> T stringToBean(String str, Class<T> clazz) {
+ if(str == null || str.length() <= 0 || clazz == null) {
+ return null;
+ }
+ if(clazz == int.class || clazz == Integer.class) {
+ return (T)Integer.valueOf(str);
+ }else if(clazz == String.class) {
+ return (T)str;
+ }else if(clazz == long.class || clazz == Long.class) {
+ return (T)Long.valueOf(str);
+ }else {
+ return JSON.toJavaObject(JSON.parseObject(str), clazz);
+ }
+ }
+
+ private void returnToPool(Jedis jedis) {
+ if(jedis != null) {
+ jedis.close();
+ }
+ }
+
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/redis/TaskKey.java b/src/main/java/org/springblade/modules/rabbitmq/redis/TaskKey.java
new file mode 100644
index 0000000..75892d7
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/redis/TaskKey.java
@@ -0,0 +1,10 @@
+package org.springblade.modules.rabbitmq.redis;
+
+public class TaskKey extends BasePrefix {
+
+ private TaskKey(int expireSeconds, String prefix) {
+ super(expireSeconds, prefix);
+ }
+
+ public static TaskKey getTaskStock= new TaskKey(0, "task");
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/vo/Man.java b/src/main/java/org/springblade/modules/rabbitmq/vo/Man.java
new file mode 100644
index 0000000..aa037aa
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/vo/Man.java
@@ -0,0 +1,16 @@
+package org.springblade.modules.rabbitmq.vo;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 人
+ */
+@Data
+public class Man implements Serializable {
+
+ private Long id;
+
+ private String username;
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/vo/Task.java b/src/main/java/org/springblade/modules/rabbitmq/vo/Task.java
new file mode 100644
index 0000000..7251761
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/vo/Task.java
@@ -0,0 +1,21 @@
+package org.springblade.modules.rabbitmq.vo;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 任务
+ */
+@Data
+public class Task implements Serializable {
+
+ private Long id;
+
+ private String taskName;
+
+ /**
+ * 数量
+ */
+ private Integer num;
+}
diff --git a/src/main/java/org/springblade/modules/rabbitmq/vo/TaskMessage.java b/src/main/java/org/springblade/modules/rabbitmq/vo/TaskMessage.java
new file mode 100644
index 0000000..6c56d45
--- /dev/null
+++ b/src/main/java/org/springblade/modules/rabbitmq/vo/TaskMessage.java
@@ -0,0 +1,13 @@
+package org.springblade.modules.rabbitmq.vo;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class TaskMessage implements Serializable {
+
+ private Long manId;
+
+ private Long taskId;
+}
diff --git a/src/main/resources/application-test.yml b/src/main/resources/application-test.yml
index 8a39539..b6f5851 100644
--- a/src/main/resources/application-test.yml
+++ b/src/main/resources/application-test.yml
@@ -13,10 +13,15 @@
# commandTimeout: 5000
datasource:
# MySql
- url: jdbc:mysql://61.131.136.25:2083/zhbaw?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&serverTimezone=GMT%2B8&nullCatalogMeansCurrent=true&allowPublicKeyRetrieval=true
+ url: jdbc:mysql://192.168.0.191:3306/zhbaw?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&serverTimezone=GMT%2B8&nullCatalogMeansCurrent=true&allowPublicKeyRetrieval=true
username: root
- password: ZHba@0112
-
+ password: root
+ # rabbitmq 设置
+# rabbitmq:
+# host: 192.168.0.191
+# port: 5672
+# username: admin
+# password: admin
# PostgreSQL
#url: jdbc:postgresql://127.0.0.1:5432/bladex_boot
#username: postgres
@@ -30,6 +35,16 @@
#username: sa
#password: zhbaw@2021
+#jedis 配置
+#redis:
+# ##redis 单机环境配置
+# host: 192.168.0.191
+# port: 6379
+# password:
+# timeout: 10
+# poolMaxTotal: 1000
+# poolMaxIdle: 500
+# poolMaxWait: 500
#图片批量上传zip
upload:
--
Gitblit v1.9.3