高并发多线程 – 产品学

Java Web框架缓存 Cache ReentrantReadWriteLock 定时器 TimerTask

1、通过ReentrantReadWriteLock、ConcurrentHashMap实现一个简易缓存。

2、通过TimerTask或Thread回收缓存。阅读全文

Java Web框架压力测试栅栏 CyclicBarrier

package mvc.test;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// 压力测试
public class StressTest {

  public static void main(String[] args) {
    
    // 并发线程数
    int count = 5;
    // 创建一个线程池
    ExecutorService exec = Executors.newCachedThreadPool();
    // 创建栅栏
    CyclicBarrier cb = new CyclicBarrier(count);
    
    // 模拟客户端访问
    for (int i = 1; i <= count; i++) {
      final int NO = i;
      Runnable run = new Runnable() {
        public void run() {
          try {
            // 等待
            cb.await();
            System.out.println("Thread:" + NO);
            String url_action = "http://localhost:8080/api/user_edit?user_id=" + String.valueOf(NO);
            System.out.println(url_action);
            
            URL url = new URL(url_action);
            HttpURLConnection connection = (HttpURLConnection) url.openConnection();
            connection.setDoInput(true);
            
            BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
            String line = "";
            String result = "";
            while ((line = in.readLine()) != null) {
              result += line;
            }
            
            System.out.println("第：" + NO + " 个");
            System.out.println(result);
                        
          } catch (Exception e) {
            e.printStackTrace();
          }
        }
      };
      
      // 执行任务
      exec.execute(run);
    }
                // 关闭线程池
    exec.shutdown();
  }

}

Java 多线程 ConcurrentHashMap

package thread.map;

import java.util.concurrent.ConcurrentHashMap;

public class Consumer implements Runnable {
  private String name;
  private ConcurrentHashMap<String, String> map;
  private int fromIndex;

  public Consumer(String name, ConcurrentHashMap<String, String> map, int fromIndex) {
    this.name = name;
    this.map = map;
    this.fromIndex = fromIndex;
  }

  public void run() {
    for (int i = fromIndex; i < fromIndex + 20; i++) {
      String str = String.valueOf(i);
      map.remove(str);
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    System.out.println(name + ", remove. size:" + map.size());
  }
}

package thread.map;

import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;

public class View implements Runnable {
  private String name;
  private ConcurrentHashMap<String, String> map;

  public View(String name, ConcurrentHashMap<String, String> map) {
    this.name = name;
    this.map = map;
  }

  public void run() {
    Iterator<String> iter = map.values().iterator();
    while (iter.hasNext()) {
      String str = iter.next();
      System.out.println(name + ", " + str);
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
}

package thread.map;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {
  /**
   * @author jzh add 2012-01-07
   * @param args
   */
  public static void main(String[] args) {
    // 初始ConcurrentHashMap
    ConcurrentHashMap<String, String> map = new ConcurrentHashMap<String, String>();
    for (int i = 0; i < 100; i++) {
      String str = String.valueOf(i);
      map.put(str, str);
        }
    
    // 创建一个线程池(创建一个可根据需要创建新线程的线程池)
    // Executors 类提供了用于此包中所提供的执行程序服务的工厂方法
    ExecutorService pool = Executors.newCachedThreadPool();
    
    Consumer consumer0 = new Consumer("howsky0", map, 10);
    Consumer consumer1 = new Consumer("howsky1", map, 40);
    Consumer consumer2 = new Consumer("howsky2", map, 70);
    View view = new View("jiang", map);

    pool.execute(consumer0);
    pool.execute(consumer1);
    pool.execute(consumer2);
    pool.execute(view);
    
    pool.shutdown();

  }
}

Java 多线程 CountDownLatch ConcurrentLinkedQueue

package thread.linkedqueue;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;

public class Worker implements Runnable {
  private CountDownLatch downLatch;
  private String name;
  private ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<Integer>();

  public Worker(CountDownLatch downLatch, String name, ConcurrentLinkedQueue<Integer> queue) {
    this.downLatch = downLatch;
    this.name = name;
    this.queue = queue;
  }

  public void run() {
    this.doPoll();
    this.downLatch.countDown();
  }

  /**
   * 取数据
   */
  private void doPoll() {
    while (!queue.isEmpty()) {
      System.out.println(name + "," + queue.poll());
    }
  }

}

package thread.linkedqueue;

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

public class Test {
  /**
   * @author jzh add 2012-01-06
   * @param args
   */
  @SuppressWarnings("unused")
  public static void main(String[] args) {
    // 初始队列
    ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<Integer>();
    for (int i = 0; i < 100000; i++) {  
            queue.offer(i);  
        }
    
    // 用于计算时间
    long timer = System.currentTimeMillis();
    
    // 创建一个线程池(创建一个可根据需要创建新线程的线程池)
    // Executors 类提供了用于此包中所提供的执行程序服务的工厂方法
    ExecutorService pool = Executors.newCachedThreadPool();

    // 初始化一个4次的CountDownLatch
    CountDownLatch latch = new CountDownLatch(4);
    
    Worker wroker1 = new Worker(latch, "chanpinxue.cn", queue);
    Worker wroker2 = new Worker(latch, "jiang", queue);
    Worker wroker3 = new Worker(latch, "jzh", queue);
    Worker wroker4 = new Worker(latch, "jiangzhihao", queue);

    pool.execute(wroker1);
    pool.execute(wroker2);
    pool.execute(wroker3);
    pool.execute(wroker4);

    try {
      latch.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    // 输出耗时
    System.out.println("time is " + (System.currentTimeMillis() - timer) + " ms");
    
    pool.shutdown();

  }
}

Java 多线程 ScheduledExecutorService

package thread.schedule;

/**
 * 实现Runnable接口
 * @author add by jzh 2011-12-28
 *
 */
public class Hello implements Runnable {
  private String name;

  public Hello(String name) {
    this.name = name;
  }

  @Override
  public void run() {
    System.out.println(name);
  }
}

package thread.schedule;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Test {
  /**
   * @author jzh add 2011-12-28
   * @param args
   */
  public static void main(String[] args) {

    // 创建一个线程池(指定线程数量为3)
    // Executors 类提供了用于此包中所提供的执行程序服务的工厂方法
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(3);

    // 创建并执行在给定延迟后启用的一次性操作。
    pool.schedule(new Hello("chanpinxue.cn"), 1, TimeUnit.SECONDS);
    // 创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；
    // 也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行，
    // 接着在 initialDelay + 2 * period 后执行，依此类推。
    pool.scheduleAtFixedRate(new Hello("jiangzhihao"), 2, 1, TimeUnit.SECONDS);
    
    // 创建并执行一个在给定初始延迟后首次启用的定期操作，
    // 随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。
    pool.scheduleWithFixedDelay(new Hello("jzh"), 3, 2, TimeUnit.SECONDS);

    try {
      Thread.sleep(10000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    
    pool.shutdown();
  }
}

Java 多线程 ExecutorService Semaphore

package thread.semaphore;

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

public class Test {
  
  /**
   * @author jzh 2011-12-27
   * @param args
   */
  public static void main(String[] args) {

    // 创建一个线程池(创建一个可根据需要创建新线程的线程池)
    // Executors 类提供了用于此包中所提供的执行程序服务的工厂方法
    ExecutorService pool = Executors.newCachedThreadPool();
    
    // 设置许可数（5个柜台）
    final Semaphore semaphore = new Semaphore(5);
    // 设置客户端数量（10个顾客）
    for (int i = 1; i <= 10; i++) {
      final int code = i;
      Runnable run = new Runnable() {
        public void run() {
          try {
            // 从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
            semaphore.acquire();
            System.out.println("第" + code + "个顾客开始获得服务");
            Thread.sleep(300);
            // 释放一个许可，将其返回给信号量。
            semaphore.release();
            System.out.println("信号量中当前可用的许可数: "+ semaphore.availablePermits());

          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      };
      pool.execute(run);
    }
    
    // 退出线程池
    pool.shutdown();
  }

}

Java 多线程 CountDownLatch

package thread.latch;

import java.util.concurrent.CountDownLatch;

public class Leader implements Runnable {
  private CountDownLatch downLatch;

  public Leader(CountDownLatch downLatch) {
    this.downLatch = downLatch;
  }

  public void run() {
    System.out.println("leader 正在等所有的worker把活干完。");
    try {
      this.downLatch.await();
    } catch (InterruptedException e) {
    }
    System.out.println("worker 把活都干完了，leader 开始检查了。");
  }
}

package thread.latch;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class Worker implements Runnable {
  private CountDownLatch downLatch;
  private String name;
  private String workContent;

  public Worker(CountDownLatch downLatch, String name, String workContent) {
    this.downLatch = downLatch;
    this.name = name;
    this.workContent = workContent;
  }

  public void run() {
    this.doSomething();
    try {
      TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException ie) {
    }
    System.out.println(this.name + " " + workContent + " 做完了");
    this.downLatch.countDown();
  }

  private void doSomething() {
    System.out.println(this.name + " 正在 " + workContent);
  }

}

package thread.latch;

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

public class Test {
  
  /**
   * @author jzh add 2011-12-25
   * @param args
   */
  public static void main(String[] args) {
    // 创建一个线程池(创建一个可根据需要创建新线程的线程池)
    // Executors 类提供了用于此包中所提供的执行程序服务的工厂方法
    ExecutorService pool = Executors.newCachedThreadPool();

    // 初始化一个4次的CountDownLatch
    CountDownLatch latch = new CountDownLatch(4);
    
    Worker wroker1 = new Worker(latch, "chanpinxue.cn", "扫地");
    Worker wroker2 = new Worker(latch, "jiang", "擦桌子");
    Worker wroker3 = new Worker(latch, "jzh", "洗碗");
    Worker wroker4 = new Worker(latch, "jiangzhihao", "关窗");

    Leader leader = new Leader(latch);

    pool.execute(wroker1);
    pool.execute(wroker2);
    pool.execute(wroker3);
    pool.execute(wroker4);

    pool.execute(leader);

    pool.shutdown();

  }

}

Java 多线程 ThreadGroup

package thread.group;

public class SayThread extends Thread {
  
  public SayThread(ThreadGroup group, String name) {
    super(group, name);
  }

  public void run() {
    for (int i = 0; i < 5; i++) {
      System.out.println(getName() + " = " + i);
      try {
        sleep(200);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
    System.out.println(getName() + " done");
  }
}

package thread.group;

public class Test {
  /**
   * @author jzh add 2011-12-20
   * @param args
   */
  public static void main(String args[]) {
    ThreadGroup group = new ThreadGroup("groupThread");
    SayThread howskyThread = new SayThread(group, "jiang");
        SayThread jzhThread = new SayThread(group, "jzh");
    SayThread sayThread = new SayThread(group, "say");
    
    howskyThread.start();
    jzhThread.start();
    sayThread.start();

    System.out.println("the number of active threads = " + group.activeCount());

    // Copies into the specified array every active thread in this thread group and its subgroups.
    Thread[] list = new Thread[group.activeCount()];
    group.enumerate(list);

    for (Thread thread : list) {
      System.out.println(thread.getName());
    }
  }
}

Java 多线程 ExecutorService Exchanger

package thread.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

  /**
   * @author jzh 2011-12-27
   * @param args
   */
  public static void main(String[] args) {
    // 创建一个线程池(创建一个可根据需要创建新线程的线程池)
    // Executors 类提供了用于此包中所提供的执行程序服务的工厂方法
    ExecutorService pool = Executors.newCachedThreadPool();
    
    final Exchanger<String> exchanger = new Exchanger<String>();
    
    // 交换者
    pool.execute(new Runnable() {
      public void run() {
        try {
          String put = "iphone";
          System.out.println("jiang" + "正在把" + put + "换出去");
          Thread.sleep(100);
          String take = (String) exchanger.exchange(put);
          System.out.println("jiang"+ "换回" + take);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
    
    // 交换者
    pool.execute(new Runnable() {
      public void run() {
        try {
          String put = "ipad";
          System.out.println("jzh" + "正在把" + put + "换出去");
          Thread.sleep(200);
          String take = (String) exchanger.exchange(put);
          System.out.println("jzh" + "换回" + take);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
    
    // 退出线程池
    pool.shutdown();
    
  }

}

Java 多线程生产者消费者仓库

package thread.control;

public class ConsumerThread extends Thread {
  private Warehouse warehouse;
  private int number;

  public ConsumerThread(Warehouse obj, int number) {
    warehouse = obj;
    this.number = number;
  }

  public void run() {
    int value = 0;
    for (int i = 0; i < 10; i++) {
      value = warehouse.get();
      System.out.println("Consumer #" + this.number + " got: " + value);
    }
  }
}

package thread.control;

public class ProducerConsumerTest {
  
  /**
   * @author jzh 2011-12-19
   * @param args
   */
  public static void main(String[] args) {
    Warehouse warehouse = new Warehouse();
    ProducerThread producerThread = new ProducerThread(warehouse, 1);
    ConsumerThread consumerThread = new ConsumerThread(warehouse, 1);
    producerThread.start();
    consumerThread.start();
  }
}

package thread.control;

public class ProducerThread extends Thread {
  private Warehouse warehouse;
  private int number;

  public ProducerThread(Warehouse obj, int number) {
    warehouse = obj;
    this.number = number;
  }

  public void run() {
    for (int i = 0; i < 10; i++) {
      warehouse.put(i);
      System.out.println("Producer #" + this.number + " put: " + i);
      try {
        sleep(100);
      } catch (InterruptedException e) {
      }
    }
  }
}

package thread.control;

public class Warehouse {
  private int content;
  private boolean available = false;

  public synchronized void put(int value) {
    while (available == true) {
      try {
        wait();
      } catch (InterruptedException e) {
      }
    }
    content = value;
    available = true;
    notifyAll();
  }

  public synchronized int get() {
    while (available == false) {
      try {
        wait();
      } catch (InterruptedException e) {
      }
    }
    available = false;
    notifyAll();
    return content;
  }

}