并发编程与线程安全

项目放在码云上托管 地址为https://gitee.com/alanwang123/concurrency
下载到本地 git clone https://gitee.com/alanwang123/concurrency.git
创建springboot项目

git status查询目录状态
git add. 将没有被版本管理的文件添加上来
git commit -am 'init' 初始提交代码
git push 提交到远程仓库

定义注解

package com.alan.concurrency.annoations;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * 通过注解标识，哪些类是【线程安全】的。
 */

//这里的作用目标配置为TYPE,可以是类和接口使用
@Target(ElementType.TYPE)
//使用SOURCE.编译的时候会去掉，这里只是方便我们的代码标注
@Retention(RetentionPolicy.SOURCE)
public @interface ThreadSafe {

    String value() default "";

}

并发模拟

图片.png

• 导入lombok包，方法测试和生成bean.（生成bean用注解即可 @data）

使用postman进行测试

图片.png

• 配置模拟并发

  
  图片.png

• 运行结果

  
  

Apache Bench(AB)

ab -n 1000 -c 50 http://localhost:8080/test
-n 测试请求数量 1000 -c 指定并发的请求数 50

Concurrency Level:      50
Time taken for tests:   0.298 seconds
Complete requests:      1000
Failed requests:        0
Total transferred:      136000 bytes
HTML transferred:       4000 bytes
Requests per second:    3350.93 [#/sec] (mean)
Time per request:       14.921 [ms] (mean)  //用户等待时间
Time per request:       0.298 [ms] (mean, across all concurrent requests)
Transfer rate:          445.04 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    5   2.5      5      15
Processing:     0   10  11.1      5      54
Waiting:        0    8  10.7      4      51
Total:          4   15  10.6     10      57

Jmeter

• 配置线程组

  
  图片.png

• 设置http请求

  
  图片.png

• 运行结果

  
  图片.png

图片.png

Semaphore CountDownLatch 进行测试

package com.alan.concurrency;


import com.alan.concurrency.annoations.NotThreadSafe;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

@Slf4j
@NotThreadSafe
public class ConcurrencyTest {


    //请求数1000
    public static int clientTotal = 5000;
    //同时并发执行的线程数
    public static int threadTotal = 200;

    public static int count = 0;


    private static void add(){
        count++;
    }

    public static void main(String[] args) throws InterruptedException {

        //定义线程池ExecutorService接口
        ExecutorService executorService = Executors.newCachedThreadPool();
        //定义信号量,传入并发线程数 final修饰不允许重新赋值
        final Semaphore semaphore = new Semaphore(threadTotal);
        //定义计数器闭锁。传入请求总数
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);

        for (int i = 0; i < clientTotal; i++) {
            //通过匿名内部类方式
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //semaphore控制并发数量
                        semaphore.acquire();
                        add();
                        semaphore.release();
                    } catch (InterruptedException e) {
                        log.error("exception",e);
                    }
                    //每次执行计数器减掉一个
                    countDownLatch.countDown();
                }

            });

        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}",count);
    }
}