精华内容
下载资源
问答
  • 主要为大家详细介绍了Java线程间通信方式,以代码结合文字的方式来讨论线程的通信,感兴趣的朋友可以参考一下
  • java线程间通信方式

    2018-10-13 17:41:11
    本质上就是共享内存式的通信,这个共享内存在java的内存模型中就是主内存,相当于通过主内存的数据进行线程通信。因Synchronized解锁时会将工作内存中的数据刷新到主内存中,Synchronized加锁时会将工作内存中的值...

    1.同步

    就是通过Synchronized关键字来进行同步访问控制,确保谁拿到了相应的锁才能执行相应的操作

    本质上就是共享内存式的通信,这个共享内存在java的内存模型中就是主内存,相当于通过主内存的数据进行线程通信。因Synchronized解锁时会将工作内存中的数据刷新到主内存中,Synchronized加锁时会将工作内存中的值清空从主内存读。多个线程访问同一变量,谁拿到了锁谁就去访问。

    2.while轮询

    假设我们添加线程向一个List中存入元素(一直存入),判断线程判断如果存入的元素达到了3个,我们就退出存入元素的线程,转而进入另一个等待它存入三个元素的线程。那么这个等待的线程我们使用while轮询list集合中是否达到了三个,如果到了三个我们就进行下一步,没有就一直轮询,

    问题:

    1.我们发现判断的线程如果没达到它的要求,cpu执行到它时就一直空转,白白浪费

    2.这个方法还有一个问题,我们没办法保证可见性,也就是说假设当加入元素的线程到达了3个,但是此时元素数量3只是处于工作内存中,那么在它将工作内存中的3刷新到共享内存中的这段时间中可能又加入了新的元素,加入后等待线程取到值,可是这时候已经大于3了,所以程序会出现问题。

    3.wait/notify机制

    这是一个Object里的方法,两个方法的作用就是沉睡和唤醒,当我们的等待线程发现没有达到想要的条件我们就沉睡它,此时另一个线程来加入元素,当元素数量达到了3 ,我们可以唤醒等待线程,告诉他你的条件达到了,你继续执行吧

    问题:如果说添加元素的线程一下添加了3个,进行了唤醒操作,但是等待线程还没运行到wait,这时产生了次空唤醒。当等待线程执行到wait之后沉睡,因为它要依靠添加元素的线程唤醒,但是添加元素的线程已经进行了唤醒,因此会一直沉睡。

    4.消息管道:

    就是通过一条管道传输线程之间通信的消息。

    展开全文
  • 下面小编就为大家带来一篇深入理解JAVA线程线程间通信方式。小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
  • java线程间通讯的几种方式

    万次阅读 多人点赞 2018-11-06 20:28:29
    这里用流在两个线程间通信,但是Java中的Stream是单向的,所以在两个线程中分别建了一个input和output public class PipedDemo { private final PipedInputStream inputStream1 ; private final...

    一、使用同一个共享变量控制

    Synchronized、wait、notify

    public class Demo1 {
    
        private final List<Integer> list =new ArrayList<>();
    
        public static void main(String[] args) {
            Demo1 demo =new Demo1();
            new Thread(()->{
                for (int i=0;i<10;i++){
                    synchronized (demo.list){
                        if(demo.list.size()%2==1){
                            try {
                                demo.list.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        demo.list.add(i);
                        System.out.print(Thread.currentThread().getName());
                        System.out.println(demo.list);
                        demo.list.notify();
                    }
                }
    
            }).start();
    
            new Thread(()->{
                for (int i=0;i<10;i++){
                    synchronized (demo.list){
                        if(demo.list.size()%2==0){
                            try {
                                demo.list.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        demo.list.add(i);
                        System.out.print(Thread.currentThread().getName());
                        System.out.println(demo.list);
                        demo.list.notify();
                    }
                }
            }).start();
        }
    }
    

    Lock、Condition

    public class Task {
        private final Lock lock = new ReentrantLock();
    
        private final Condition addConditon = lock.newCondition();
        private final Condition subConditon = lock.newCondition();
    
        private volatile int num = 0;
        private List<String> list = new ArrayList<>();
    
        public void add() {
            for (int i = 0; i < 10; i++) {
                lock.lock();
    
                try {
                    if (list.size() == 10) {
                        addConditon.await();
                    }
                    num++;
                    Thread.sleep(100);
                    list.add("add " + num);
                    System.out.println("The list size is " + list.size());
                    System.out.println("The add thread is " + Thread.currentThread().getName());
                    System.out.println("-------------");
                    subConditon.signal();
    
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }
    
        public void sub() {
            for (int i = 0; i < 10; i++) {
                lock.lock();
    
                try {
    
                    if (list.size() == 0) {
                        subConditon.await();
                    }
                    num--;
                    Thread.sleep(100);
                    list.remove(0);
                    System.out.println("The list size is " + list.size());
                    System.out.println("The sub thread is " + Thread.currentThread().getName());
                    System.out.println("-------------");
                    addConditon.signal();
    
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }
    
    
        public static void main(String[] args) {
            Task task = new Task();
            new Thread(task::add).start();
            new Thread(task::sub).start();
        }
    }
    

    利用volatile

    volatile修饰的变量值直接存在主内存里面,子线程对该变量的读写直接写住内存,而不是像其它变量一样在local thread里面产生一份copy。volatile能保证所修饰的变量对于多个线程可见性,即只要被修改,其它线程读到的一定是最新的值。

    public class Demo2 {
        private volatile List<Integer> list =new ArrayList<>();
        public static void main(String[] args) {
            Demo2 demo =new Demo2();
            new Thread(()->{
                for (int i=0;i<10;i++){
                        demo.list.add(i);
                        System.out.print(Thread.currentThread().getName());
                        System.out.println(demo.list);
                }
    
            }).start();
    
            new Thread(()->{
                for (int i=0;i<10;i++){
                        demo.list.add(i);
                        System.out.print(Thread.currentThread().getName());
                        System.out.println(demo.list);
                    }
            }).start();
        }
    }
    

    利用AtomicInteger

    和volatile类似

    二、PipedInputStream、PipedOutputStream

    这里用流在两个线程间通信,但是Java中的Stream是单向的,所以在两个线程中分别建了一个input和output

    public class PipedDemo {
    
        private final PipedInputStream inputStream1;
        private final PipedOutputStream outputStream1;
        private final PipedInputStream inputStream2;
        private final PipedOutputStream outputStream2;
    
        public PipedDemo(){
            inputStream1 = new PipedInputStream();
            outputStream1 = new PipedOutputStream();
            inputStream2 = new PipedInputStream();
            outputStream2 = new PipedOutputStream();
            try {
                inputStream1.connect(outputStream2);
                inputStream2.connect(outputStream1);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
        /**程序退出时,需要关闭stream*/
        public void shutdown() throws IOException {
            inputStream1.close();
            inputStream2.close();
            outputStream1.close();
            outputStream2.close();
        }
    
    
        public static void main(String[] args) throws IOException {
            PipedDemo demo =new PipedDemo();
            new Thread(()->{
                PipedInputStream in = demo.inputStream2;
                PipedOutputStream out = demo.outputStream2;
    
                for (int i = 0; i < 10; i++) {
                    try {
                        byte[] inArr = new byte[2];
                        in.read(inArr);
                        System.out.print(Thread.currentThread().getName()+": "+i+" ");
                        System.out.println(new String(inArr));
                        while(true){
                            if("go".equals(new String(inArr)))
                                break;
                        }
                        out.write("ok".getBytes());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
    
            }).start();
    
            new Thread(()->{
                PipedInputStream in = demo.inputStream1;
                PipedOutputStream out = demo.outputStream1;
    
                for (int i = 0; i < 10; i++) {
                    try {
                        out.write("go".getBytes());
                        byte[] inArr = new byte[2];
                        in.read(inArr);
                        System.out.print(Thread.currentThread().getName()+": "+i+" ");
                        System.out.println(new String(inArr));
                        while(true){
                            if("ok".equals(new String(inArr)))
                                break;
                        }
    
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
    
            }).start();
    //        demo.shutdown();
        }
    }
    

    输出

    Thread-0: 0 go
    Thread-1: 0 ok
    Thread-0: 1 go
    Thread-1: 1 ok
    Thread-0: 2 go
    Thread-1: 2 ok
    Thread-0: 3 go
    Thread-1: 3 ok
    Thread-0: 4 go
    Thread-1: 4 ok
    Thread-0: 5 go
    Thread-1: 5 ok
    Thread-0: 6 go
    Thread-1: 6 ok
    Thread-0: 7 go
    Thread-1: 7 ok
    Thread-0: 8 go
    Thread-1: 8 ok
    Thread-0: 9 go
    Thread-1: 9 ok
    

    三、利用BlockingQueue

    BlockingQueue定义的常用方法如下:

    • add(Object):把Object加到BlockingQueue里,如果BlockingQueue可以容纳,则返回true,否则抛出异常。
    • offer(Object):表示如果可能的话,将Object加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false。
    • put(Object):把Object加到BlockingQueue里,如果BlockingQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里有空间再继续。
    • poll(time):获取并删除BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null。当不传入time值时,立刻返回。
    • peek():立刻获取BlockingQueue里排在首位的对象,但不从队列里删除,如果队列为空,则返回null。
    • take():获取并删除BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的对象被加入为止。

    BlockingQueue有四个具体的实现类:

    • ArrayBlockingQueue:数组阻塞队列,规定大小,其构造函数必须带一个int参数来指明其大小。其所含的对象是以FIFO(先入先出)顺序排序的。
    • LinkedBlockingQueue:链阻塞队列,大小不定,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定。其所含的对象是以FIFO顺序排序的。
    • PriorityBlockingQueue:类似于LinkedBlockingQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数所带的Comparator决定的顺序。
    • SynchronousQueue:特殊的BlockingQueue,它的内部同时只能够容纳单个元素,对其的操作必须是放和取交替完成的。
    • DelayQueue:延迟队列,注入其中的元素必须实现 java.util.concurrent.Delayed 接口

    所有BlockingQueue的使用方式类似,以下例子一个线程写入,一个线程读取,操作的是同一个Queue:

    public class BlockingQueueDemo {
    
        public static void main(String[] args) {
            LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
            //读线程
            new Thread(() -> {
                int i =0;
                while (true) {
                    try {
                        String item = queue.take();
                        System.out.print(Thread.currentThread().getName() + ": " + i + " ");
                        System.out.println(item);
                        i++;
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
            //写线程
            new Thread(() -> {
                for (int i = 0; i < 10; i++) {
                    try {
                        String item = "go"+i;
                        System.out.print(Thread.currentThread().getName() + ": " + i + " ");
                        System.out.println(item);
                        queue.put(item);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
    
    展开全文
  • JAVA线程间通信的几种方式

    万次阅读 多人点赞 2017-08-12 21:55:12
    “编写两个线程,一个线程打印1~25,另一个线程打印字母A~Z,打印顺序为12A34B56C……5152Z,要求使用线程间通信。” 这是一道非常好的面试题,非常能彰显被面者关于多线程的功力,一下子就勾起了我的兴趣。这里...
    今天在群里面看到一个很有意思的面试题:
    
    “编写两个线程,一个线程打印1~25,另一个线程打印字母A~Z,打印顺序为12A34B56C……5152Z,要求使用线程间的通信。”

    这是一道非常好的面试题,非常能彰显被面者关于多线程的功力,一下子就勾起了我的兴趣。这里抛砖引玉,给出7种想到的解法。

    通用代码:

    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * Created by Edison Xu on 2017/3/2.
     */
    public enum Helper {
    
        instance;
    
        private static final ExecutorService tPool = Executors.newFixedThreadPool(2);
    
        public static String[] buildNoArr(int max) {
            String[] noArr = new String[max];
            for(int i=0;i<max;i++){
                noArr[i] = Integer.toString(i+1);
            }
            return noArr;
        }
    
        public static String[] buildCharArr(int max) {
            String[] charArr = new String[max];
            int tmp = 65;
            for(int i=0;i<max;i++){
                charArr[i] = String.valueOf((char)(tmp+i));
            }
            return charArr;
        }
    
        public static void print(String... input){
            if(input==null)
                return;
            for(String each:input){
                System.out.print(each);
            }
        }
    
        public void run(Runnable r){
            tPool.submit(r);
        }
    
        public void shutdown(){
            tPool.shutdown();
        }
    
    }

    1. 第一种解法,包含多种小的不同实现方式,但一个共同点就是靠一个共享变量来做控制

    a. 利用最基本的synchronizednotifywait

    public class MethodOne {
        private final ThreadToGo threadToGo = new ThreadToGo();
        public Runnable newThreadOne() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    try {
                        for (int i = 0; i < arr.length; i=i+2) {
                            synchronized (threadToGo) {
                                while (threadToGo.value == 2)
                                    threadToGo.wait();
                                Helper.print(arr[i], arr[i + 1]);
                                threadToGo.value = 2;
                                threadToGo.notify();
                            }
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Oops...");
                    }
                }
            };
        }
        public Runnable newThreadTwo() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    try {
                        for (int i = 0; i < arr.length; i++) {
                            synchronized (threadToGo) {
                                while (threadToGo.value == 1)
                                    threadToGo.wait();
                                Helper.print(arr[i]);
                                threadToGo.value = 1;
                                threadToGo.notify();
                            }
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Oops...");
                    }
                }
            };
        }
        class ThreadToGo {
            int value = 1;
        }
        public static void main(String args[]) throws InterruptedException {
            MethodOne one = new MethodOne();
            Helper.instance.run(one.newThreadOne());
            Helper.instance.run(one.newThreadTwo());
            Helper.instance.shutdown();
        }
    }

    b. 利用LockCondition

    public class MethodTwo {
        private Lock lock = new ReentrantLock(true);
        private Condition condition = lock.newCondition();
        private final ThreadToGo threadToGo = new ThreadToGo();
        public Runnable newThreadOne() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i=i+2) {
                        try {
                            lock.lock();
                            while(threadToGo.value == 2)
                                condition.await();
                            Helper.print(arr[i], arr[i + 1]);
                            threadToGo.value = 2;
                            condition.signal();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } finally {
                            lock.unlock();
                        }
                    }
                }
            };
        }
        public Runnable newThreadTwo() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i++) {
                        try {
                            lock.lock();
                            while(threadToGo.value == 1)
                                condition.await();
                            Helper.print(arr[i]);
                            threadToGo.value = 1;
                            condition.signal();
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            lock.unlock();
                        }
                    }
                }
            };
        }
        class ThreadToGo {
            int value = 1;
        }
        public static void main(String args[]) throws InterruptedException {
            MethodTwo two = new MethodTwo();
            Helper.instance.run(two.newThreadOne());
            Helper.instance.run(two.newThreadTwo());
            Helper.instance.shutdown();
        }
    }
    c. 利用volatile:
    volatile 修饰的变量值直接存在main memory里面,子线程对该变量的读写直接写入main memory,而不是像其它变量一样在local thread里面产生一份copy。 volatile 能保证所修饰的变量对于多个线程可见性,即只要被修改,其它线程读到的一定是最新的值。
    public class MethodThree {
        private volatile ThreadToGo threadToGo = new ThreadToGo();
        class ThreadToGo {
            int value = 1;
        }
        public Runnable newThreadOne() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i=i+2) {
                        while(threadToGo.value==2){}
                        Helper.print(arr[i], arr[i + 1]);
                        threadToGo.value=2;
                    }
                }
            };
        }
        public Runnable newThreadTwo() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i++) {
                        while(threadToGo.value==1){}
                        Helper.print(arr[i]);
                        threadToGo.value=1;
                    }
                }
            };
        }
        public static void main(String args[]) throws InterruptedException {
            MethodThree three = new MethodThree();
            Helper.instance.run(three.newThreadOne());
            Helper.instance.run(three.newThreadTwo());
            Helper.instance.shutdown();
        }
    }

    d. 利用AtomicInteger

    public class MethodFive {
        private AtomicInteger threadToGo = new AtomicInteger(1);
        public Runnable newThreadOne() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i=i+2) {
                        while(threadToGo.get()==2){}
                        Helper.print(arr[i], arr[i + 1]);
                        threadToGo.set(2);
                    }
                }
            };
        }
        public Runnable newThreadTwo() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i++) {
                        while(threadToGo.get()==1){}
                        Helper.print(arr[i]);
                        threadToGo.set(1);
                    }
                }
            };
        }
        public static void main(String args[]) throws InterruptedException {
            MethodFive five = new MethodFive();
            Helper.instance.run(five.newThreadOne());
            Helper.instance.run(five.newThreadTwo());
            Helper.instance.shutdown();
        }
    }

    2. 第二种解法,是利用CyclicBarrierAPI;

    CyclicBarrier可以实现让一组线程在全部到达Barrier时(执行await()),再一起同时执行,并且所有线程释放后,还能复用它,即为Cyclic。
    CyclicBarrier类提供两个构造器:

    public CyclicBarrier(int parties, Runnable barrierAction) {
    }
    public CyclicBarrier(int parties) {
    }
    public class MethodFour{
          private final CyclicBarrier barrier;
          private final List<String> list;
          public MethodFour() {
              list = Collections.synchronizedList(new ArrayList<String>());
              barrier = new CyclicBarrier(2,newBarrierAction());
          }
          public Runnable newThreadOne() {
              final String[] inputArr = Helper.buildNoArr(52);
              return new Runnable() {
                  private String[] arr = inputArr;
                  public void run() {
                      for (int i = 0, j=0; i < arr.length; i=i+2,j++) {
                          try {
                              list.add(arr[i]);
                              list.add(arr[i+1]);
                              barrier.await();
                          } catch (InterruptedException | BrokenBarrierException e) {
                              e.printStackTrace();
                          }
                      }
                  }
              };
          }
          public Runnable newThreadTwo() {
              final String[] inputArr = Helper.buildCharArr(26);
              return new Runnable() {
                  private String[] arr = inputArr;
                  public void run() {
                      for (int i = 0; i < arr.length; i++) {
                          try {
                              list.add(arr[i]);
                              barrier.await();
                          } catch (InterruptedException | BrokenBarrierException e) {
                              e.printStackTrace();
                          }
                      }
                  }
              };
          }
          private Runnable newBarrierAction(){
              return new Runnable() {
                  @Override
                  public void run() {
                      Collections.sort(list);
                      list.forEach(c->System.out.print(c));
                      list.clear();
                  }
              };
          }
          public static void main(String args[]){
              MethodFour four = new MethodFour();
              Helper.instance.run(four.newThreadOne());
              Helper.instance.run(four.newThreadTwo());
              Helper.instance.shutdown();
          }
    }
    这里多说一点,这个API其实还是利用lockcondition,无非是多个线程去争抢CyclicBarrier的instance的lock罢了,最终barrierAction执行时,是在抢到CyclicBarrierinstance的那个线程上执行的。

    3. 第三种解法,是利用PipedInputStreamAPI;

    这里用流在两个线程间通信,但是Java中的Stream是单向的,所以在两个线程中分别建了一个input和output。这显然是一种很搓的方式,不过也算是一种通信方式吧……-_-T,执行的时候那种速度简直。。。请不要BS我。
    public class MethodSix {
        private final PipedInputStream inputStream1;
        private final PipedOutputStream outputStream1;
        private final PipedInputStream inputStream2;
        private final PipedOutputStream outputStream2;
        private final byte[] MSG;
        public MethodSix() {
            inputStream1 = new PipedInputStream();
            outputStream1 = new PipedOutputStream();
            inputStream2 = new PipedInputStream();
            outputStream2 = new PipedOutputStream();
            MSG = "Go".getBytes();
            try {
                inputStream1.connect(outputStream2);
                inputStream2.connect(outputStream1);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        public void shutdown() throws IOException {
            inputStream1.close();
            inputStream2.close();
            outputStream1.close();
            outputStream2.close();
        }
        public Runnable newThreadOne() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                private PipedInputStream in = inputStream1;
                private PipedOutputStream out = outputStream1;
                public void run() {
                    for (int i = 0; i < arr.length; i=i+2) {
                        Helper.print(arr[i], arr[i + 1]);
                        try {
                            out.write(MSG);
                            byte[] inArr = new byte[2];
                            in.read(inArr);
                            while(true){
                                if("Go".equals(new String(inArr)))
                                    break;
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
        }
        public Runnable newThreadTwo() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                private PipedInputStream in = inputStream2;
                private PipedOutputStream out = outputStream2;
                public void run() {
                    for (int i = 0; i < arr.length; i++) {
                        try {
                            byte[] inArr = new byte[2];
                            in.read(inArr);
                            while(true){
                                if("Go".equals(new String(inArr)))
                                    break;
                            }
                            Helper.print(arr[i]);
                            out.write(MSG);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
        }
        public static void main(String args[]) throws IOException {
            MethodSix six = new MethodSix();
            Helper.instance.run(six.newThreadOne());
            Helper.instance.run(six.newThreadTwo());
            Helper.instance.shutdown();
            six.shutdown();
        }

    4. 第四种解法,是利用BlockingQueue

    顺便总结下BlockingQueue的一些内容。
    BlockingQueue定义的常用方法如下:

    • add(Object):把Object加到BlockingQueue里,如果BlockingQueue可以容纳,则返回true,否则抛出异常。
    • offer(Object):表示如果可能的话,将Object加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false。
    • put(Object):把Object加到BlockingQueue里,如果BlockingQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里有空间再继续。
    • poll(time):获取并删除BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null。当不传入time值时,立刻返回。
    • peek():立刻获取BlockingQueue里排在首位的对象,但不从队列里删除,如果队列为空,则返回null。
    • take():获取并删除BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的对象被加入为止。

    BlockingQueue有四个具体的实现类:

    • ArrayBlockingQueue:规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小。其所含的对象是以FIFO(先入先出)顺序排序的。
    • LinkedBlockingQueue:大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定。其所含的对象是以FIFO顺序排序的。
    • PriorityBlockingQueue:类似于LinkedBlockingQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数所带的Comparator决定的顺序。
    • SynchronousQueue:特殊的BlockingQueue,对其的操作必须是放和取交替完成的。

    这里我用了两种玩法:

    • 一种是共享一个queue,根据peekpoll的不同来实现;
    • 第二种是两个queue,利用take()会自动阻塞来实现。
    public class MethodSeven {
        private final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
        public Runnable newThreadOne() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i=i+2) {
                        Helper.print(arr[i], arr[i + 1]);
                        queue.offer("TwoToGo");
                        while(!"OneToGo".equals(queue.peek())){}
                        queue.poll();
                    }
                }
            };
        }
        public Runnable newThreadTwo() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i++) {
                        while(!"TwoToGo".equals(queue.peek())){}
                        queue.poll();
                        Helper.print(arr[i]);
                        queue.offer("OneToGo");
                    }
                }
            };
        }
        private final LinkedBlockingQueue<String> queue1 = new LinkedBlockingQueue<>();
        private final LinkedBlockingQueue<String> queue2 = new LinkedBlockingQueue<>();
        public Runnable newThreadThree() {
            final String[] inputArr = Helper.buildNoArr(52);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i=i+2) {
                        Helper.print(arr[i], arr[i + 1]);
                        try {
                            queue2.put("TwoToGo");
                            queue1.take();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
        }
        public Runnable newThreadFour() {
            final String[] inputArr = Helper.buildCharArr(26);
            return new Runnable() {
                private String[] arr = inputArr;
                public void run() {
                    for (int i = 0; i < arr.length; i++) {
                        try {
                            queue2.take();
                            Helper.print(arr[i]);
                            queue1.put("OneToGo");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
        }
        public static void main(String args[]) throws InterruptedException {
            MethodSeven seven = new MethodSeven();
            Helper.instance.run(seven.newThreadOne());
            Helper.instance.run(seven.newThreadTwo());
            Thread.sleep(2000);
            System.out.println("");
            Helper.instance.run(seven.newThreadThree());
            Helper.instance.run(seven.newThreadFour());
            Helper.instance.shutdown();
        }

    原文链接: http://edisonxu.org/2017/03/02/java-thread-communication.html






    展开全文
  • java线程间通信方式

    千次阅读 2020-12-12 20:46:07
    java线程间通信方式1. 共享变量2. 等待/通知3. 管道流 1. 共享变量 volatile修饰的变量,线程可见,可使用这种变量作为线程传递消息的媒介; 延伸出来的,还有redis中的值,数据库中的值,都可以作为线程...

    java线程间通信的方式

    1. 共享变量

    • volatile修饰的变量,线程间可见,可使用这种变量作为线程间传递消息的媒介;
    • 延伸出来的,还有redis中的值,数据库中的值,都可以作为线程间共同访问的变量;

    2. 等待/通知

    • 同步代码中利用锁对象的wait和notify方法来进行通信;
    • 经典案例如thread.join()方法,里面就是用等待通知机制实现的;

    3. 管道流

    • 管道输入/输出流和普通的文件输入/输出流或者网络输入/输出流不同之处在于,它主要用于线程之间的数据传输,而传输的媒介为内存。
    • 管道输入/输出流主要包括了如下4种具体实现:PipedOutputStream、PipedInputStream、PipedReader和PipedWriter,前两种面向字节,而后两种面向字符。
    package com.demo.other;
    
    import java.io.PipedReader;
    import java.io.PipedWriter;
    
    public class Piped {
        public static void main(String[] args) throws Exception {
            PipedWriter out = new PipedWriter();
            PipedReader in = new PipedReader();
    // 将输出流和输入流进行连接,否则在使用时会抛出IOException
            out.connect(in);
            Thread printThread = new Thread(new Print(in), "PrintThread");
            printThread.start();
            int receive = 0;
            try {
                while ((receive = System.in.read()) != -1) {
                    out.write(receive);
                }
            } finally {
                out.close();
            }
        }
    
    
        static class Print implements Runnable {
            private PipedReader in;
            public Print(PipedReader in) {
                this.in = in;
            }
    
            @Override
            public void run() {
                int receive = 0;
                try {
                    while ((receive = in.read()) != -1) {
                        System.out.print((char) receive);
                    }
                } catch (Exception ex) {
                }
            }
        }
    }
    

    打印:

    hello Kitty
    hello Kitty
    

    通过管道流可以将数据从一个线程写入,而另一个线程读出;

    展开全文
  • java 线程间通信的几种方式

    千次阅读 2017-10-11 14:12:12
    假设有两个线程,一个线程A,一个线程B,两个线程分别依次打印 1-3 三个数字即可。package Test;/** /** * @author Administrator * @createdate 2017-10-10 */ public class demo1 { public static void main...
  • java实现线程通信的几种方式

    万次阅读 2020-05-30 19:34:35
    比如大家熟知的消息中间件的实现,从某种角度上讲,就借助了多线程通信的思想,下面总结了JDK中常用的几种实现线程通信方式,提供参考 1、synchronized实现方式 可能很多小伙伴们会有疑问,synchronized是对共享...
  • Java线程之间的通信方式

    千次阅读 2020-08-19 09:47:01
    对于线程通信来说,线程同步可以归纳为线程通信的一个子集,对于线程通信指的是两个线程之间可以交换一些实时的数据信息,而线程同步只交换一些控制信息。 1 synchronized package Communication; class ...
  • 原文:https://blog.csdn.net/Hadwin1991/article/details/73527835本文主要针对JAVA线程线程之间的通信方式进行分析解释,主要以代码结合文字的方式来讨论线程间的通信。synchronized同步public class MyObject...
  • java线程同步与线程间通信

    千次阅读 2019-07-17 11:25:52
    java线程同步和通信的方法有如下几种: 1、synchronized关键字修饰方法或代码段,实现数据的互斥访问 2、volatile修饰变量,实现多线程环境下数据的同步 3、ReentrantLock可重入锁,实现数据的互斥访问 3、...
  • 本文将介绍常用的线程间通信工具CountDownLatch、CyclicBarrier和Phaser的用法,并结合实例介绍它们各自的适用场景及相同点和不同点。
  • java线程通信的三种方式

    千次阅读 2019-04-02 14:26:08
    1、传统的线程通信。 在synchronized修饰的同步方法或者修饰的同步代码块中使用Object类提供的wait(),notify()和notifyAll()3个方法进行线程通信。 关于这3个方法的解释: wait():导致当前线程等待,直到其他线程...
  • 主要介绍了java线程间通信的通俗解释,介绍了线程通信中的几个相关概念,然后分享了线程通信的实现方式及代码示例,具有一定参考价值 ,需要的朋友可以了解下。
  • 线程同步是线程之间按照⼀定的顺序执⾏,可以使⽤锁来实现达到线程同步,也就是在需要同步的代码块里加上关键字synchronized 。 二、信号量 共享内存机制 基于 volatile 关键字来实现的 volitile关键字能够保证内存...
  • 线程间通讯方式有哪些? 一、进程通信方式 管道( pipe ): 管道是一种半双工的通信方式,数据只能单向流动,而且只能在具有亲缘关系的进程使用。进程的亲缘关系通常是指父子进程关系。 有名管道 (namedpipe) ...
  • Java中的多线程线程间通信

    万次阅读 2018-09-23 10:22:42
    线程间通信: /学习笔记/ 多个线程在处理同一资源,但是任务却不同。 先看一个例子,采用两个线程执行进行输入和输出任务: //资源 class Resource { String name; String sex; } //输入 ...
  • JAVA线程通信详解

    万次阅读 多人点赞 2018-10-14 09:11:36
    五、线程间通信——管道 六、方法Join的使用 一、概述  线程线程之间不是相互独立的个体,它们彼此之间需要相互通信和协作,最典型的例子就是生产者-消费者问题:当队列满时,生产者需要等待队列有空间才能...
  • JAVA线程线程间通信方式

    千次阅读 2018-12-22 23:56:02
    一,介绍本总结我对于JAVA线程线程之间的通信方式的理解,主要以代码结合文字的方式来讨论线程间的通信,故摘抄了书中的一些示例代码。二,线程间通信方式①同步这里讲的同...
  • 主要介绍了Java线程通信实现方式详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
  • 进程之间通信方式 (1) 管道(PIPE) (2) 命名管道(FIFO) (3) 信号量(Semphore) (4) 消息队列(MessageQueue) (5) 共享内存(SharedMemory) (6) Socket Java如何支持进程通信。我们把Java进程理解为JVM进程。很明显,...
  • 一、进程间通信(IPC,Inter-Process Communication)是指在不同进程传播或交换信息 1. 无名管道 特点 半双工(数据流向仅有一个方向),具有固定的读端和写端 只能用于父进程或兄弟线程之间通信(具有血缘关系的...
  • 线程间通讯方式

    千次阅读 2019-06-04 16:09:41
    线程间通讯方式 1:同步(synchronized) 2:wait/notify()机制 3:管道通信就是使用java.io.PipedInputStream 和 java.io.PipedOutputStream进行通信 进程间通讯方式 ... ...
  • Java线程通信详解

    2020-09-01 08:37:44
    本篇文章主要介绍了Java线程通信问题,线程通信用来保证线程协调运行,有需要的朋友可以了解一下。
  • Java线程之间通信方式

    2019-12-31 11:02:23
    通信方式: 1.同步 a.同步是指多个线程之间通过synchronize关键字这种方式来实现线程间通讯。 b.这种方式本质上就是共享内存式的通讯。多个线程需要访问同一个共享变量,谁拿到了锁(获得了访问权限),谁就...
  • java线程间通信

    万次阅读 2019-04-10 13:39:44
    在前面我们将了很多关于同步的问题,然而在现实中,需要线程之间的协作。比如说最经典的生产者-消费者模型:当队列满时,生产者需要等待队列有空间才能继续往里面放入商品,而在等待的期间内,生产者必须释放对临界...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 219,317
精华内容 87,726
关键字:

java线程间的通信方式

java 订阅