ArrayBlockingQueue简单介绍

ArrayBlockingQueue是一个用数组实现的环形队列，在构造方法中，会要求传入数组的容量

创建数组长度为10的有界队列，数据类型为字符串
BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10);

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.java</groupId>
    <artifactId>test-study</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/>
    </parent>

    <dependencies>
        <!--tomcat容器-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--lombok依赖-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.16</version>
        </dependency>
        <!--引入junit单元测试依赖-->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <!--判断空的用法  -->
        <dependency>
            <groupId>commons-lang</groupId>
            <artifactId>commons-lang</artifactId>
            <version>2.6</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8 -->
        <dependency>
            <groupId>com.oracle.database.jdbc</groupId>
            <artifactId>ojdbc8</artifactId>
            <version>12.2.0.1</version>
        </dependency>
        <!--springboot整合mybatis-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>2.1.2</version>
        </dependency>

        <!--添加fastjson依赖-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.70</version>
        </dependency>
        <!-- 热部署模块 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <optional>true</optional> <!-- 这个需要为 true 热部署才有效 -->
        </dependency>
        <!--ThreadFactoryBuilder的依赖包,多线程使用-->
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>30.1-jre</version>
        </dependency>


    </dependencies>


    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
        <finalName>study</finalName>
    </build>


</project>

server.port=2001


logging.level.com.java.test=debug
logging.level.web=debug
spring.devtools.add-properties=false

package com.java.test;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @Description:
 * @Author: Yourheart
 * @Create: 2022/10/20 15:32
 */
@SpringBootApplication
public class TestApplication {
    public static void main(String[] args) {
        SpringApplication.run(TestApplication.class,args);
    }
}

　创建线程池，调用位置

package com.java.test.blockingqueuedemo;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.*;

/**
 * @Description:
 * @Author: Yourheart
 * @Create: 2022/11/7 14:22
 */
@Slf4j
@RestController
public class ArrayBlockingQueueDemo {

    BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10);


    ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
            .setNameFormat("demo-pool-%d").build();


    ExecutorService singleThreadPool = new ThreadPoolExecutor(2, 10,
            0L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(10), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());


    /**
     * 启动线程放置数据
     * @return
     */
    @GetMapping("/blockingQueueTest")
    public String blockingQueueTest() {

        singleThreadPool.execute(new ArrayBlockingQueueThread(blockingQueue));

        return blockingQueue.poll();

    }

    /**
     * 启动线程取出数据
     * @return
     */
    @GetMapping("/blockingQueueGet")
    public String blockingQueueGet(long threadSleepTime) {
        singleThreadPool.execute(new ArrayBlockingQueueGet(blockingQueue,threadSleepTime));
        blockingQueue.forEach(a->{
            log.info("队列中的数据:{}",a.toString());
        });
        return blockingQueue.poll();
    }

    @GetMapping("/blockingQueueQuery")
    public String blockingQueueQuery(long threadSleepTime) {
        singleThreadPool.execute(new ArrayBlockingQueueGet(blockingQueue,threadSleepTime));
        blockingQueue.forEach(a->{
            log.info("队列中的数据:{}",a.toString());
        });
        return blockingQueue.poll();
    }
}

http://127.0.0.1:2001/blockingQueueTest

http://127.0.0.1:2001/blockingQueueGet?threadSleepTime=500

http://127.0.0.1:2001/blockingQueueQuery?threadSleepTime=500

package com.java.test.blockingqueuedemo;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.util.concurrent.BlockingQueue;

/**
 * @Description:
 * @Author: Yourheart
 * @Create: 2022/11/7 17:04
 */
@Slf4j
public class ArrayBlockingQueueGet implements Runnable {

    private BlockingQueue<String> blockingQueue;

    private long threadSleepTime;


    public ArrayBlockingQueueGet(BlockingQueue<String> blockingQueue, long threadSleepTime) {
        this.blockingQueue = blockingQueue;
        this.threadSleepTime = threadSleepTime;
    }

    @Override
    public void run() {
        while(true){
            synchronized (this){
                try {
                    String poll = blockingQueue.poll();
                    if (StringUtils.isBlank(poll)){
                        continue;
                    }
                    log.info("poll:{}",poll);
                    //将此处的睡眠时间分别改为100和1000，观察运行结果
                    Thread.sleep(threadSleepTime);
                    blockingQueue.take();
                    log.info("队列中剩余数据个数:{}",blockingQueue.size());
                    blockingQueue.forEach(a->{
                        log.info("【队列中剩余的数据】:{}",a.toString());
                    });
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

package com.java.test.blockingqueuedemo;

import lombok.extern.slf4j.Slf4j;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BlockingQueue;

/**
 * @Description:
 * @Author: Yourheart
 * @Create: 2022/11/7 16:52
 */
@Slf4j
public class ArrayBlockingQueueThread implements Runnable {

    BlockingQueue<String> blockingQueue;

    public ArrayBlockingQueueThread(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        while (true){
            try {
                Thread.sleep(500);
               // String toString = UUID.randomUUID().toString();
                Date date = new Date();
                SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                String format = simpleDateFormat.format(date);
                blockingQueue.put(format);
                log.info("线程:{},放置的数据:{},当前队列中的数据个数:{}",Thread.currentThread().getName(),format,blockingQueue.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}