import java.util.concurrent.LinkedBlockingQueue;  
  
public class LinkedBlockingQueueExample {  
    public static void main(String[] args) {  
        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();  
  
        // 生產(chǎn)者線程  
        new Thread(() -> {  
            for (int i = 0; i < 10; i++) {  
                try {  
                    queue.put("Element " + i);  
                    System.out.println("Produced: Element " + i);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
        }).start();  
        // 消費者線程  
        new Thread(() -> {  
            for (int i = 0; i < 10; i++) {  
                try {  
                    String element = queue.take();  
                    System.out.println("Consumed: " + element);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
        }).start();  
    }  
}

127.0.0.1:6379> xack mq group1 1580959593553-0
(integer) 1

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-redis</artifactId>
    <version>1.4.7.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.23.1</version>
</dependency>

@Component
@Slf4j
public class BeforeQueue {
    @Autowired
    private RedissonClient redissonClient;


    /**
     * <p>取出元素</p>
     * <p>如果隊列中沒有元素，就阻塞等待，直</p>
     * @return
     */
    public Object take(){
        RBlockingQueue<Object> queue1 = redissonClient.getBlockingQueue("queue1");
        Object obj = null;
        try {
            obj = queue1.take();
            log.info("從myqueue1取出元素:{}",obj.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return obj;
    }


    /**
     * <p>放入元素</p>
     * @param obj
     */
    public void offer(Object obj){
        RBlockingDeque<Object> queue1 = redissonClient.getBlockingDeque("queue1");
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(queue1);
        delayedQueue.offer(obj,5, TimeUnit.SECONDS);
        log.info("向myqueue1設(shè)置元素:{}",obj.toString());
    }
}

@Component
@Slf4j
public class RunningQueue {
    @Autowired
    private RedissonClient redissonClient;


    public Object take(){
        RBlockingQueue<Object> queue1 = redissonClient.getBlockingQueue("queue2");
        Object obj = null;
        try {
            obj = queue1.take();
            log.info("從myqueue2取出元素:{}",obj.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return obj;
    }


    public void offer(Object obj){
        RBlockingDeque<Object> queue1 = redissonClient.getBlockingDeque("queue2");
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(queue1);
        delayedQueue.offer(obj,5, TimeUnit.SECONDS);
        log.info("向myqueue2設(shè)置元素:{}",obj.toString());
    }
}

@Component
@Slf4j
public class CompleteQueue {
    @Autowired
    private RedissonClient redissonClient;


    public Object take(){
        RBlockingQueue<Object> queue1 = redissonClient.getBlockingQueue("queue3");
        Object obj = null;
        try {
            obj = queue1.take();
            log.info("從CompleteQueue取出元素:{}",obj.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return obj;
    }


    public void offer(Object obj){
        RBlockingQueue<Object> queue1 = redissonClient.getBlockingDeque("queue3");
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(queue1);
        delayedQueue.offer(obj,5, TimeUnit.SECONDS);
        log.info("向CompleteQueue設(shè)置元素:{}",obj.toString());
    }
}

@Component
@Slf4j
public class BeforeQueueListener implements Listener{
    @Autowired
    private BeforeQueue beforeQueue;
    @Autowired
    private RunningQueue runningQueue;
    @Override
    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    log.info("監(jiān)聽器進(jìn)入阻塞:BeforeQueueListener");
                    Object obj = beforeQueue.take();
                    if (ObjectUtil.isNotNull(obj)) {
                        try {
                            log.info("開始休眠1s模擬業(yè)務(wù)處理:BeforeQueueListener,元素:{}",obj.toString());
                            Thread.currentThread().sleep(1000);
                            log.info("業(yè)務(wù)處理完成:BeforeQueueListener,元素:{}",obj.toString());
                            runningQueue.offer(obj);
                        } catch (InterruptedException e) {
                            log.error("業(yè)務(wù)處理發(fā)生異常,重置元素到BeforeQueue隊列中");
                            log.error(e.getMessage());
                            beforeQueue.offer(obj);
                        }


                    }
                }
            }
        }).start();
    }
}

@Component
@Slf4j
public class RunningQueueListener implements Listener {


    @Autowired
    private RunningQueue runningQueue;
    @Autowired
    private CompleteQueue completeQueue;


    @Override
    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    log.info("監(jiān)聽器進(jìn)入阻塞:RunningQueueListener");
                    Object obj = runningQueue.take();
                    if (ObjectUtil.isNotNull(obj)) {
                        try {
                            log.info("開始休眠1s模擬業(yè)務(wù)處理:RunningQueueListener,元素:{}", obj.toString());
                            Thread.currentThread().sleep(1000);
                            Random random = new Random();
                            int i = random.nextInt(2);
                            if (i==0) {
                                test();
                            }
                            log.info("業(yè)務(wù)處理完成:RunningQueueListener,元素:{}", obj.toString());
                            completeQueue.offer(obj);
                        } catch (Exception e) {
                            log.error("業(yè)務(wù)處理發(fā)生異常,重置元素到RunningQueue隊列中");
                            log.error(e.getMessage());
                            runningQueue.offer(obj);
                        }
                    }
                }
            }
        }).start();
    }


    public void test(){
        try {
            int i=1/0;
        } catch (Exception e) {
           throw  new RuntimeException("除數(shù)異常");
        }
    }


}

@Component
@Slf4j
public class CompleteQueueListener implements Listener{


    @Autowired
    private CompleteQueue completeQueue;
    @Override
    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    log.info("監(jiān)聽器進(jìn)入阻塞:CompleteQueueListener");
                    Object obj = completeQueue.take();
                    if (ObjectUtil.isNotNull(obj)) {
                        try {
                            log.info("開始休眠1s模擬業(yè)務(wù)處理:CompleteQueueListener,元素:{}",obj.toString());
                            Thread.currentThread().sleep(1000);
                            log.info("業(yè)務(wù)處理完成:listener3,元素:{}",obj.toString());
                        } catch (InterruptedException e) {
                            log.error("業(yè)務(wù)處理發(fā)生異常,重置元素到CompleteQueue隊列中");
                            log.error(e.getMessage());
                            completeQueue.offer(obj);
                        }
                       log.info("CompleteQueueListener任務(wù)結(jié)束,元素:{}",obj.toString());
                    }
                }
            }
        }).start();
    }
}

@Component
public class MyRunner implements ApplicationRunner {
    @Autowired
    private ApplicationContext applicationContext;
    @Override
    public void run(ApplicationArguments args) throws Exception {
        Map<String, Listener> beansOfType = applicationContext.getBeansOfType(Listener.class);
        for (String s : beansOfType.keySet()) {
            Listener listener = beansOfType.get(s);
            listener.start();
        }


    }
}

<?xml versinotallow="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
    <!-- 日志存放路徑 -->
    <property name="log.path" value="logs/"/>
    <!-- 日志輸出格式 -->
    <property name="console.log.pattern"
              value="%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight(%-5level) %magenta(${PID:-}) - %green([%-21thread]) %cyan(%-35logger{30}) %msg%n"/>
    <!-- 控制臺輸出 -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${console.log.pattern}</pattern>
            <charset>utf-8</charset>
        </encoder>
    </appender>
    <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
        <queueSize>500</queueSize>
        <discardingThreshold>0</discardingThreshold>
        <neverBlock>true</neverBlock>
        <appender-ref ref="console" />
    </appender>
    <!--系統(tǒng)操作日志-->
    <root level="info">
        <appender-ref ref="ASYNC" />
    </root>
</configuration>