devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

多线程执行_多线程执行程序

作者：Monodyee | 2024-03-06 10:04:20

赞

踩

多线程执行程序

一、交互执行abc

1、队列+park+volatile

  1.     private static LinkedList<Thread> list = new LinkedList<>();
  2.     private static volatile Boolean state = new Boolean(false);
  3.  
  4.     public static void main(String[] args) {
  5.         Thread thread1 = new Thread(() -> {
  6.             while (true) {
  7.                 LockSupport.park();
  8.                 System.out.println("a");
  9.                 state = new Boolean(true);
  10.             }
  11.         });
  12.         thread1.start();
  13.         list.add(thread1);
  14.         
  15.         Thread thread2 = new Thread(() -> {
  16.             while (true) {
  17.                 LockSupport.park();
  18.                 System.out.println("b");
  19.                 state = new Boolean(true);
  20.             }
  21.         });
  22.         thread2.start();
  23.         list.add(thread2);
  24.  
  25.         Thread thread3 = new Thread(() -> {
  26.             while (true) {
  27.                 LockSupport.park();
  28.                 System.out.println("c");
  29.                 state = new Boolean(true);
  30.             }
  31.         });
  32.         thread3.start();
  33.         list.add(thread3);
  34.  
  35.         while (true) {
  36.             Thread take = list.poll();
  37.             LockSupport.unpark(take);
  38.             while (true) {
  39.                 if (state.booleanValue() == true) {
  40.                     state = new Boolean(false);
  41.                     list.add(take);
  42.                     break;
  43.                 }
  44.             }
  45.         }
  46.     }

二、一个打印偶数一个打印奇数

1、synchronized

  1.     private static final Object lock = new Object();
  2.     private static int num = 0;
  3.  
  4.     public static void main(String[] args) throws InterruptedException {
  5.         Thread thread1 = new Thread(() -> {
  6.             while (true) {
  7.                 synchronized (lock) {
  8.                     try {
  9.                         System.out.println(Thread.currentThread().getName() + "=" + num++);
  10.                         lock.notifyAll();
  11.                         lock.wait();
  12.                     } catch (InterruptedException e) {
  13.                         e.printStackTrace();
  14.                     }
  15.                 }
  16.             }
  17.         });
  18.  
  19.  
  20.         Thread thread2 = new Thread(() -> {
  21.             while (true) {
  22.                 synchronized (lock) {
  23.                     try {
  24.                         System.out.println(Thread.currentThread().getName() + "=" + num++);
  25.                         lock.notifyAll();
  26.                         lock.wait();
  27.                     } catch (InterruptedException e) {
  28.                         e.printStackTrace();
  29.                     }
  30.                 }
  31.             }
  32.         });
  33.         thread1.start();
  34.         thread2.start();

2、Exchanger（两个线程无序）

  1. private static Exchanger<Integer> exchanger = new Exchanger<>();
  2.  
  3.     public static void main(String[] args) throws InterruptedException {
  4.         Thread thread1 = new Thread(() -> {
  5.             Integer num = -1;
  6.             while (true) {
  7.                 try {
  8.                     num = num + 2;
  9.                     Integer exchange = exchanger.exchange(num);
  10.                     System.out.println(Thread.currentThread().getName() + "=" + exchange);
  11.                 } catch (InterruptedException e) {
  12.                     e.printStackTrace();
  13.                 }
  14.             }
  15.         });
  16.  
  17.  
  18.         Thread thread2 = new Thread(() -> {
  19.             Integer num = -2;
  20.             while (true) {
  21.                 try {
  22.                     num = num + 2;
  23.                     Integer exchange = exchanger.exchange(num);
  24.                     System.out.println(Thread.currentThread().getName() + "=" + exchange);
  25.                 } catch (InterruptedException e) {
  26.                     e.printStackTrace();
  27.                 }
  28.             }
  29.         });
  30.         thread1.start();
  31.         thread2.start();
  32.  
  33.  
  34.     }

三、按照顺序，三个线程分别打印A5次，B10次，C15次

1、synchronized

  1. private static int flag = 1;
  2.     private static Object lock = new Object();
  3.  
  4.     public static void main(String[] args) {
  5.  
  6.  
  7.         Thread thread1 = new Thread(() -> {
  8.             synchronized (lock) {
  9.                 while (true) {
  10.                     try {
  11.                         if (flag == 1) {
  12.                             for (int i = 0; i < 5; i++) {
  13.                                 System.out.println("A");
  14.                             }
  15.                             flag++;
  16.                         }
  17.                         lock.notifyAll();
  18.                         lock.wait();
  19.                     } catch (InterruptedException e) {
  20.                         e.printStackTrace();
  21.                     }
  22.                 }
  23.             }
  24.         });
  25.  
  26.         Thread thread2 = new Thread(() -> {
  27.             synchronized (lock) {
  28.                 while (true) {
  29.                     try {
  30.                         if (flag == 2) {
  31.                             for (int i = 0; i < 10; i++) {
  32.                                 System.out.println("B");
  33.                             }
  34.                             flag++;
  35.                         }
  36.                         lock.notifyAll();
  37.                         lock.wait();
  38.                     } catch (InterruptedException e) {
  39.                         e.printStackTrace();
  40.                     }
  41.                 }
  42.             }
  43.         });
  44.  
  45.         Thread thread3 = new Thread(() -> {
  46.             synchronized (lock) {
  47.                 while (true) {
  48.                     try {
  49.                         if (flag == 3) {
  50.                             for (int i = 0; i < 15; i++) {
  51.                                 System.out.println("C");
  52.                             }
  53.                             flag = 1;
  54.                         }
  55.                         lock.notifyAll();
  56.                         lock.wait();
  57.                     } catch (InterruptedException e) {
  58.                         e.printStackTrace();
  59.                     }
  60.                 }
  61.             }
  62.         });
  63.         thread1.start();
  64.         thread2.start();
  65.         thread3.start();
  66.     }

2、lock

  1. static volatile int flag = 1;
  2.  
  3.     public static void main(String[] args) {
  4.  
  5.         Lock lock = new ReentrantLock();
  6.         Condition c1 = lock.newCondition();
  7.         Condition c2 = lock.newCondition();
  8.         Condition c3 = lock.newCondition();
  9.  
  10.  
  11.         Thread thread1 = new Thread(() -> {
  12.             while (true) {
  13.                 lock.lock();
  14.                 try {
  15.                     if (flag != 1) {
  16.                         c1.await();
  17.                     }
  18.                     for (int i = 0; i < 5; i++) {
  19.                         System.out.println("A");
  20.                     }
  21.                     flag = 2;
  22.                     c2.signal();
  23.                 } catch (Exception e) {
  24.                     e.printStackTrace();
  25.                 } finally {
  26.                     lock.unlock();
  27.                 }
  28.             }
  29.         });
  30.  
  31.         Thread thread2 = new Thread(() -> {
  32.             while (true) {
  33.                 lock.lock();
  34.                 try {
  35.                     if (flag != 2) {
  36.                         c2.await();
  37.                     }
  38.                     for (int i = 0; i < 10; i++) {
  39.                         System.out.println("B");
  40.                     }
  41.                     flag = 3;
  42.                     c3.signal();
  43.                 } catch (Exception e) {
  44.                     e.printStackTrace();
  45.                 } finally {
  46.                     lock.unlock();
  47.                 }
  48.             }
  49.         });
  50.  
  51.         Thread thread3 = new Thread(() -> {
  52.             while (true) {
  53.                 lock.lock();
  54.                 try {
  55.                     if (flag != 3) {
  56.                         c3.await();
  57.                     }
  58.                     for (int i = 0; i < 15; i++) {
  59.                         System.out.println("C");
  60.                     }
  61.                     flag = 1;
  62.                     c1.signal();
  63.                 } catch (Exception e) {
  64.                     e.printStackTrace();
  65.                 } finally {
  66.                     lock.unlock();
  67.                 }
  68.             }
  69.         });
  70.         thread1.start();
  71.         thread2.start();
  72.         thread3.start();
  73.     }

四、创建水分子

  1.  private static Lock lock = new ReentrantLock();
  2.     private static Condition c1 = lock.newCondition();
  3.     private static Condition c2 = lock.newCondition();
  4.     private static int num = 1;
  5.  
  6.     public static void main(String[] args) {
  7.         Thread thread1 = new Thread(() -> {
  8.             while (true) {
  9.                 lock.lock();
  10.                 try {
  11.                     if (num == 3) {
  12.                         c2.signal();
  13.                         c1.await();
  14.                     }
  15.                     System.out.print("H" + "--------" + num);
  16.                     num++;
  17.                 } catch (InterruptedException e) {
  18.                     e.printStackTrace();
  19.                 } finally {
  20.                     lock.unlock();
  21.                 }
  22.             }
  23.         });
  24.         Thread thread2 = new Thread(() -> {
  25.             while (true) {
  26.                 lock.lock();
  27.                 try {
  28.                     if (num != 3) {
  29.                         c1.signal();
  30.                         c2.await();
  31.                     }
  32.                     System.out.println("O" + "--------" + num);
  33.                     num = 1;
  34.                 } catch (InterruptedException e) {
  35.                     e.printStackTrace();
  36.                 } finally {
  37.                     lock.unlock();
  38.                 }
  39.             }
  40.         });
  41.         thread1.start();
  42.         thread2.start();
  43.     }

五、生产者，消费者

1、Disruptor

  1. public static void main(String[] args) {
  2.         EventFactory<Message> factory = new MessageFactory();
  3.  
  4.         int ringBufferSize = 1024 * 1024;
  5.  
  6.         Disruptor<Message> disruptor =
  7.  
  8.                 new Disruptor<Message>(factory, ringBufferSize, Executors.defaultThreadFactory(), ProducerType.SINGLE, new YieldingWaitStrategy());
  9.  
  10.         disruptor.handleEventsWith(new MessageHandler("handler-1")).then(new MessageHandler("handler-2"), new MessageHandler("handler-3"));
  11.  
  12.         disruptor.start();
  13.  
  14.         RingBuffer<Message> ringBuffer = disruptor.getRingBuffer();
  15.  
  16.         Producer producer = new Producer(ringBuffer);
  17.  
  18.         //单生产者，生产3条数据
  19.  
  20.         for (int l = 0; l < 3; l++) {
  21.  
  22.             producer.onData(l + "");
  23.  
  24.         }
  25.  
  26.  
  27.         disruptor.shutdown();
  28.  
  29.     }
  30.  
  31.     static class Producer {
  32.  
  33.         private RingBuffer<Message> ringBuffer;
  34.  
  35.         public Producer(RingBuffer<Message> ringBuffer) {
  36.             this.ringBuffer = ringBuffer;
  37.         }
  38.  
  39.         public void onData(String text) {
  40.             long next = this.ringBuffer.next();
  41.             Message message = ringBuffer.get(next);
  42.             message.setId(UUID.randomUUID().toString());
  43.             message.setText("信息" + text);
  44.             ringBuffer.publish(next);
  45.         }
  46.  
  47.     }
  48.  
  49.  
  50.     static class MessageHandler implements EventHandler<Message> {
  51.  
  52.         String handlerName;
  53.  
  54.         public MessageHandler(String handlerName) {
  55.             this.handlerName = handlerName;
  56.         }
  57.  
  58.         @Override
  59.         public void onEvent(Message message, long l, boolean b) {
  60.             message.addHandler(handlerName + "已经处理");
  61.             System.out.println("消息:" + message.getText() + "  " + handlerName + "已经处理");
  62.         }
  63.     }
  64.  
  65.     static class MessageFactory implements EventFactory<Message> {
  66.  
  67.         @Override
  68.         public Message newInstance() {
  69.             return new Message();
  70.         }
  71.     }
  72.  
  73.     static class Message {
  74.         String id;
  75.         String text;
  76.         List<String> handler = new ArrayList<>();
  77.  
  78.         public String getId() {
  79.             return id;
  80.         }
  81.  
  82.         public void setId(String id) {
  83.             this.id = id;
  84.         }
  85.  
  86.         public String getText() {
  87.             return text;
  88.         }
  89.  
  90.         public void setText(String text) {
  91.             this.text = text;
  92.         }
  93.  
  94.         public List<String> getHandler() {
  95.             return handler;
  96.         }
  97.  
  98.         public void setHandler(List<String> handler) {
  99.             this.handler = handler;
  100.         }
  101.  
  102.         public void addHandler(String handler) {
  103.             this.handler.add(handler);
  104.         }
  105.     }

2、LinkedBlockingQueue

  1. static Random random = new Random();
  2.     static LinkedBlockingQueue<String> linkedBlockingQueue = new LinkedBlockingQueue<String>(5);
  3.  
  4.     public static void main(String[] args) {
  5.         Thread thread1 = new Thread(() -> {
  6.             try {
  7.                 while (true) {
  8.                     String value = String.valueOf(random.nextInt(10));
  9.                     System.out.println("生产者1生产了——————" + value);
  10.                     linkedBlockingQueue.put(value);
  11.                 }
  12.             } catch (Exception e) {
  13.                 e.printStackTrace();
  14.             }
  15.         });
  16.  
  17.         Thread thread2 = new Thread(() -> {
  18.             try {
  19.                 while (true) {
  20.                     String take = linkedBlockingQueue.take();
  21.                     System.out.println("消费了" + take);
  22.                 }
  23.             } catch (Exception e) {
  24.                 e.printStackTrace();
  25.             }
  26.         });
  27.  
  28.         thread1.start();
  29.         thread2.start();
  30.     }

3、synchronized 

  1. static int len = 10;
  2.     static Random random = new Random();
  3.     static Queue<String> queue = new LinkedList<String>();
  4.  
  5.  
  6.     public static void main(String[] args) {
  7.         Thread thread1 = new Thread(() -> {
  8.             try {
  9.                 while (true) {
  10.                     synchronized (queue) {
  11.                         if (queue.size() == len) {
  12.                             System.out.println("当前队列满");
  13.                             queue.notifyAll();
  14.                             queue.wait();
  15.                         }
  16.                         queue.notifyAll();
  17.                         String value = String.valueOf(random.nextInt(10));
  18.                         System.out.println("生产者1生产了——————" + value);
  19.                         queue.add(value);
  20.                         Thread.yield();
  21.                     }
  22.                 }
  23.             } catch (Exception e) {
  24.                 e.printStackTrace();
  25.             }
  26.         });
  27.  
  28.         Thread thread2 = new Thread(() -> {
  29.             try {
  30.                 while (true) {
  31.                     synchronized (queue) {
  32.                         if (queue.size() == 0) {
  33.                             System.out.println("当前队列为空");
  34.                             queue.notifyAll();
  35.                             queue.wait();
  36.                         }
  37.                         queue.notifyAll();
  38.                         String take = queue.poll();
  39.                         System.out.println("消费了" + take);
  40.                         Thread.yield();
  41.                     }
  42.                 }
  43.             } catch (Exception e) {
  44.                 e.printStackTrace();
  45.             }
  46.         });
  47.  
  48.         thread1.start();
  49.         thread2.start();
  50.     }

4、Lock

  1. static int len = 10;
  2.     static Lock lock = new ReentrantLock();
  3.     static Condition condition = lock.newCondition();
  4.  
  5.     static Random random = new Random();
  6.     static Queue<String> queue = new LinkedList<>();
  7.  
  8.  
  9.     public static void main(String[] args) {
  10.         Thread thread1 = new Thread(() -> {
  11.             try {
  12.                 while (true) {
  13.                     lock.lock();
  14.                     try {
  15.                         if (queue.size() == len) {
  16.                             System.out.println("当前队列满");
  17.                             condition.await();
  18.                         }
  19.                         String value = String.valueOf(random.nextInt(10));
  20.                         System.out.println("生产者1生产了——————" + value);
  21.                         queue.add(value);
  22.                         condition.signal();
  23.                     } finally {
  24.                         lock.unlock();
  25.                     }
  26.                 }
  27.             } catch (Exception e) {
  28.                 e.printStackTrace();
  29.             }
  30.         });
  31.  
  32.         Thread thread2 = new Thread(() -> {
  33.             try {
  34.                 while (true) {
  35.                     lock.lock();
  36.                     try {
  37.                         if (queue.size() == 0) {
  38.                             System.out.println("当前队列为空");
  39.                             condition.await();
  40.                         }
  41.                         String take = queue.poll();
  42.                         System.out.println("消费了" + take);
  43.                         condition.signal();
  44.                     } finally {
  45.                         lock.unlock();
  46.                     }
  47.                 }
  48.             } catch (Exception e) {
  49.                 e.printStackTrace();
  50.             }
  51.         });
  52.  
  53.         thread1.start();
  54.         thread2.start();
  55.     }

5、Semaphore

  1. static Semaphore notFull = new Semaphore(10);
  2.     static Semaphore notEmpty = new Semaphore(0);
  3.     
  4.     static Random random = new Random();
  5.     static Queue<String> queue = new ConcurrentLinkedQueue<>();
  6.  
  7.  
  8.     public static void main(String[] args) {
  9.         Thread thread1 = new Thread(() -> {
  10.             while (true) {
  11.                 try {
  12.                     notFull.acquire();
  13.                     String value = String.valueOf(random.nextInt(10));
  14.                     System.out.println("生产者1生产了——————" + value);
  15.                     queue.add(value);
  16.                 } catch (Exception e) {
  17.                     e.printStackTrace();
  18.                 }finally {
  19.                     notEmpty.release();
  20.                 }
  21.             }
  22.         });
  23.  
  24.         Thread thread2 = new Thread(() -> {
  25.             try {
  26.                 while (true) {
  27.                     try {
  28.                         notEmpty.acquire();
  29.                         String take = queue.poll();
  30.                         System.out.println("消费了" + take);
  31.                     } catch (Exception e) {
  32.                         e.printStackTrace();
  33.                     }finally {
  34.                         notFull.release();
  35.                     }
  36.                 }
  37.             } catch (Exception e) {
  38.                 e.printStackTrace();
  39.             }
  40.         });
  41.  
  42.         thread1.start();
  43.         thread2.start();
  44.     }

6、管道

  1. static PipedOutputStream pipedOutputStream = new PipedOutputStream();
  2.     static PipedInputStream pipedInputStream = new PipedInputStream();
  3.  
  4.     static Random random = new Random();
  5.  
  6.  
  7.     public static void main(String[] args) throws IOException {
  8.         Thread thread1 = new Thread(() -> {
  9.             while (true) {
  10.                 try {
  11.                     String value = String.valueOf(random.nextInt(10));
  12.                     System.out.println("生产者1生产了——————" + value);
  13.                     pipedOutputStream.write(value.getBytes(StandardCharsets.UTF_8));
  14.                 } catch (Exception e) {
  15.                     e.printStackTrace();
  16.                 }
  17.             }
  18.         });
  19.  
  20.         Thread thread2 = new Thread(() -> {
  21.             try {
  22.                 while (true) {
  23.                     try {
  24.                         int length = -1;
  25.                         byte[] buffer = new byte[64];
  26.                         while ((length = pipedInputStream.read(buffer)) != -1) {
  27.                             System.out.println("消费了" + new String(buffer, 0, length));
  28.                         }
  29.                     } catch (Exception e) {
  30.                         e.printStackTrace();
  31.                     }
  32.                 }
  33.             } catch (Exception e) {
  34.                 e.printStackTrace();
  35.             }
  36.         });
  37.  
  38.         pipedOutputStream.connect(pipedInputStream);
  39.  
  40.         thread1.start();
  41.         thread2.start();
  42.     }

六、100个任务，分10批次，10批串行执行，10个任务并行执行

1、CyclicBarrier

批次之间结果无依赖，无返回值

  1. public class Test {
  2.  
  3.     static ExecutorService service = new ThreadPoolExecutor(200, 400, 30, TimeUnit.SECONDS,
  4.             new LinkedBlockingQueue<>(2000), new ThreadFactoryBuilder().setNameFormat("thread-%d").build());
  5.  
  6.     static CyclicBarrier cyclicBarrier = new CyclicBarrier(10);
  7.  
  8.     public static void main(String[] args) {
  9.  
  10.         List<Runnable> lists = new ArrayList<>();
  11.         for (int i = 0; i < 10; i++) {
  12.             for (int j = 0; j < 10; j++) {
  13.                 String str = "批次:" + i + " " + "任务:" + j;
  14.                 Runnable runnable = () -> {
  15.                     try {
  16.                         System.out.println(str);
  17.                         cyclicBarrier.await();
  18.                     } catch (Exception e) {
  19.                         e.printStackTrace();
  20.                     }
  21.                 };
  22.                 lists.add(runnable);
  23.             }
  24.         }
  25.         lists.forEach(service::execute);
  26.     }
  27. }

2、stream

  1. static ExecutorService executor = new ThreadPoolExecutor(200, 400, 30, TimeUnit.SECONDS,
  2.             new LinkedBlockingQueue<>(2000), new ThreadFactoryBuilder().setNameFormat("thread-%d").build());
  3.  
  4.     public static void main(String[] args) {
  5.  
  6.         List<List<Callable<String>>> lists = new ArrayList<>();
  7.         for (int i = 0; i < 10; i++) {
  8.             List<Callable<String>> list = new ArrayList<>();
  9.             for (int j = 0; j < 10; j++) {
  10.                 String str = "批次:" + i + " " + "任务:" + j;
  11.                 Callable<String> callable = () -> {
  12.                     System.out.println(str);
  13.                     return str;
  14.                 };
  15.                 list.add(callable);
  16.             }
  17.             lists.add(list);
  18.         }
  19.         List<String> listStream = lists.stream().map(s -> {
  20.             try {
  21.                 return executor.invokeAll(s);
  22.             } catch (Exception e) {
  23.                 e.printStackTrace();
  24.                 return null;
  25.             }
  26.         }).map(t -> t.parallelStream().map(fu -> {
  27.             try {
  28.                 return fu.get();
  29.             } catch (Exception e) {
  30.                 e.printStackTrace();
  31.                 return null;
  32.             }
  33.         }).collect(Collectors.toList())).flatMap(Collection::stream).collect(Collectors.toList());
  34.     }

3、CompletableFuture

  1. public static void main(String[] args) {
  2.         CompletableFuture<Void> future = new CompletableFuture<Void>();
  3.         for (int i = 0; i < 10; i++) {
  4.             CompletableFuture<Void>[] completableFutures = new CompletableFuture[10];
  5.             for (int j = 0; j < 10; j++) {
  6.                 String str = "批次:" + i + " " + "任务:" + j;
  7.                 completableFutures[j] = CompletableFuture.runAsync(() -> System.out.println(str));
  8.             }
  9.             CompletableFuture<Void> allOf = CompletableFuture.allOf(completableFutures);
  10.             future.thenRun(() -> {
  11.                 try {
  12.                     allOf.get();
  13.                 } catch (Exception e) {
  14.                     e.printStackTrace();
  15.                 }
  16.             });
  17.         }
  18.         try {
  19.             future.get();
  20.         } catch (Exception e) {
  21.             e.printStackTrace();
  22.         }
  23.     }

4、Queue

  1.     static ExecutorService executor = new ThreadPoolExecutor(200, 400, 30, TimeUnit.SECONDS,
  2.             new LinkedBlockingQueue<>(2000), new ThreadFactoryBuilder().setNameFormat("thread-%d").build());
  3.  
  4.     public static void main(String[] args) {
  5.         Queue<Queue<Callable<String>>> queues = new LinkedList<>();
  6.         for (int i = 0; i < 10; i++) {
  7.             Queue<Callable<String>> queue = new LinkedList<>();
  8.             for (int j = 0; j < 10; j++) {
  9.                 String str = "批次:" + i + " " + "任务:" + j;
  10.                 Callable<String> callable = () -> {
  11.                     System.out.println(str);
  12.                     return str;
  13.                 };
  14.                 queue.add(callable);
  15.             }
  16.             queues.add(queue);
  17.         }
  18.         try {
  19.             while (!queues.isEmpty()) {
  20.                 Queue<Callable<String>> poll = queues.poll();
  21.                 List<Future<String>> futureList = executor.invokeAll(poll);
  22.                 List<String> list = futureList.parallelStream().map(t -> {
  23.                     try {
  24.                         return t.get();
  25.                     } catch (Exception e) {
  26.                         e.printStackTrace();
  27.                         return null;
  28.                     }
  29.                 }).collect(Collectors.toList());
  30.             }
  31.         } catch (Exception e) {
  32.             e.printStackTrace();
  33.         }
  34.     }

5、lmax.disruptor

  1. public class Test15 {
  2.  
  3.     static ExecutorService service = new ThreadPoolExecutor(200, 400, 30, TimeUnit.SECONDS,
  4.             new LinkedBlockingQueue<>(2000), new ThreadFactoryBuilder().setNameFormat("thread-%d").build());
  5.  
  6.     public static void main(String[] args) {
  7.         EventFactory<Queue<Runnable>> factory = new MessageFactory();
  8.  
  9.         int ringBufferSize = 1024 * 1024;
  10.  
  11.         Disruptor<Queue<Runnable>> disruptor =
  12.  
  13.                 new Disruptor<Queue<Runnable>>(factory, ringBufferSize, Executors.defaultThreadFactory(), ProducerType.SINGLE, new YieldingWaitStrategy());
  14.         
  15.  
  16.         disruptor.handleEventsWith(new MessageHandler("handler-1"))
  17.                 .then(new MessageHandler("handler-2"))
  18.                 .then(new MessageHandler("handler-3"))
  19.                 .then(new MessageHandler("handler-4"))
  20.                 .then(new MessageHandler("handler-5"))
  21.                 .then(new MessageHandler("handler-6"))
  22.                 .then(new MessageHandler("handler-7"))
  23.                 .then(new MessageHandler("handler-8"))
  24.                 .then(new MessageHandler("handler-9"))
  25.                 .then(new MessageHandler("handler-10"));
  26.  
  27.         disruptor.start();
  28.  
  29.         RingBuffer<Queue<Runnable>> ringBuffer = disruptor.getRingBuffer();
  30.  
  31.         Producer producer = new Producer(ringBuffer);
  32.  
  33.  
  34.         Queue<Runnable> queues = new LinkedList<>();
  35.         for (int i = 0; i < 10; i++) {
  36.             for (int j = 0; j < 10; j++) {
  37.                 String str = "批次:" + i + " " + "任务:" + j;
  38.                 Runnable runnable = () -> {
  39.                     System.out.println(str);
  40.                 };
  41.                 queues.add(runnable);
  42.             }
  43.         }
  44.         producer.onData(queues);
  45.  
  46.         disruptor.shutdown();
  47.  
  48.     }
  49.  
  50.     static class Producer {
  51.  
  52.         private RingBuffer<Queue<Runnable>> ringBuffer;
  53.  
  54.         public Producer(RingBuffer<Queue<Runnable>> ringBuffer) {
  55.             this.ringBuffer = ringBuffer;
  56.         }
  57.  
  58.         public void onData(Queue<Runnable> runnable) {
  59.             long next = this.ringBuffer.next();
  60.             Queue<Runnable> runnables = ringBuffer.get(next);
  61.             runnables.addAll(runnable);
  62.             ringBuffer.publish(next);
  63.         }
  64.  
  65.     }
  66.  
  67.  
  68.     static class MessageHandler implements EventHandler<Queue<Runnable>> {
  69.  
  70.         String handlerName;
  71.  
  72.         public MessageHandler(String handlerName) {
  73.             this.handlerName = handlerName;
  74.         }
  75.  
  76.         @Override
  77.         public void onEvent(Queue<Runnable> runnables, long l, boolean b) throws Exception {
  78.             for (int i = 0; i < 10; i++) {
  79.                 service.submit(runnables.poll()).get();
  80.             }
  81.         }
  82.     }
  83.  
  84.     static class MessageFactory implements EventFactory<Queue<Runnable>> {
  85.  
  86.         @Override
  87.         public Queue<Runnable> newInstance() {
  88.             return new LinkedList<>();
  89.         }
  90.     }
  91. }

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/197738
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号