新增 rabbitmq 设置

Administrator

2022-04-12 068c5b7cbf239d6aae87fe8ae6239f1165f76431

新增 rabbitmq 设置

 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

 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>

 src/main/java/org/springblade/modules/rabbitmq/config/KeyPrefix.java

New file
@@ -0,0 +1,8 @@
package org.springblade.modules.rabbitmq.config;

public interface KeyPrefix {

    public int expireSeconds();

    public String getPrefix();
}

 src/main/java/org/springblade/modules/rabbitmq/config/MQConfig.java

New file
@@ -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);
    }
}

 src/main/java/org/springblade/modules/rabbitmq/config/MQreceiver.java

New file
@@ -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());

    }
}

 src/main/java/org/springblade/modules/rabbitmq/config/MQsender.java

New file
@@ -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");
    }

}

 src/main/java/org/springblade/modules/rabbitmq/controller/SendMessageController.java

New file
@@ -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";
    }
}

 src/main/java/org/springblade/modules/rabbitmq/redis/BasePrefix.java

New file
@@ -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;
    }

}

 src/main/java/org/springblade/modules/rabbitmq/redis/RedisConfig.java

New file
@@ -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;
    }
}

 src/main/java/org/springblade/modules/rabbitmq/redis/RedisPoolFactory.java

New file
@@ -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;
    }

}

 src/main/java/org/springblade/modules/rabbitmq/redis/RedisService.java

New file
@@ -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();
         }
    }

}

 src/main/java/org/springblade/modules/rabbitmq/redis/TaskKey.java

New file
@@ -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");
}

 src/main/java/org/springblade/modules/rabbitmq/vo/Man.java

New file
@@ -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;
}

 src/main/java/org/springblade/modules/rabbitmq/vo/Task.java

New file
@@ -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;
}

 src/main/java/org/springblade/modules/rabbitmq/vo/TaskMessage.java

New file
@@ -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;
}

 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: