在并发任务间交换数据-生产消费者3

Myth丶恋晨 2022-09-23 10:52 85阅读 0赞

## 在并发任务间交换数据 ##

Java 并发 API 提供了一种允许2个并发任务间相互交换数据的同步应用。更具体的说，Exchanger 类允许在2个线程间定义同步点，当2个线程到达这个点，他们相互交换数据类型，使用第一个线程的数据类型变成第二个的，然后第二个线程的数据类型变成第一个的。

这个类在遇到类似生产者和消费者问题时，是非常有用的。来一个非常经典的并发问题：你有相同的数据buffer，一个或多个数据生产者，和一个或多个数据消费者。只是Exchange类只能同步2个线程，所以你只能在你的生产者和消费者问题中只有一个生产者和一个消费者时使用这个类。

在这个指南，你将学习如何使用 Exchanger 类来解决只有一个生产者和一个消费者的生产者和消费者问题。

## 例子 ##

package com.packtpub.java7.concurrency.chapter3.recipe7.core;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.Exchanger;
    
    import com.packtpub.java7.concurrency.chapter3.recipe7.task.Consumer;
    import com.packtpub.java7.concurrency.chapter3.recipe7.task.Producer;
    
    /** * Main class of the example * */
    public class Main { 
    
        /** * Main method of the example * @param args */
        public static void main(String[] args) {
    
            // Creates two buffers
            List<String> buffer1=new ArrayList<>();
            List<String> buffer2=new ArrayList<>();
    
            // Creates the exchanger
            Exchanger<List<String>> exchanger=new Exchanger<>();
    
            // Creates the producer
            Producer producer=new Producer(buffer1, exchanger);
            // Creates the consumer
            Consumer consumer=new Consumer(buffer2, exchanger);
    
            // Creates and starts the threads
            Thread threadProducer=new Thread(producer);
            Thread threadConsumer=new Thread(consumer);
    
            threadProducer.start();
            threadConsumer.start();
    
        }
    
    }

package com.packtpub.java7.concurrency.chapter3.recipe7.task;
    
    import java.util.List;
    import java.util.concurrent.Exchanger;
    
    /** * This class implements the producer * */
    public class Producer implements Runnable { 
    
        /** * Buffer to save the events produced */
        private List<String> buffer;
    
        /** * Exchager to synchronize with the consumer */
        private final Exchanger<List<String>> exchanger;
    
        /** * Constructor of the class. Initializes its attributes * @param buffer Buffer to save the events produced * @param exchanger Exchanger to syncrhonize with the consumer */
        public Producer (List<String> buffer, Exchanger<List<String>> exchanger){
            this.buffer=buffer;
            this.exchanger=exchanger;
        }
    
        /** * Main method of the producer. It produces 100 events. 10 cicles of 10 events. * After produce 10 events, it uses the exchanger object to synchronize with * the consumer. The producer sends to the consumer the buffer with ten events and * receives from the consumer an empty buffer */
        @Override
        public void run() {
            int cycle=1;
    
            for (int i=0; i<10; i++){
                System.out.printf("Producer: Cycle %d\n",cycle);
    
                for (int j=0; j<10; j++){
                    String message="Event "+((i*10)+j);
                    System.out.printf("Producer: %s\n",message);
                    buffer.add(message);
                }
    
                try {
                    /* * Change the data buffer with the consumer */
                    buffer=exchanger.exchange(buffer);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
                System.out.printf("Producer: %d\n",buffer.size());
    
                cycle++;
            }
    
        }
    
    
    
    }

package com.packtpub.java7.concurrency.chapter3.recipe7.task;
    
    import java.util.List;
    import java.util.concurrent.Exchanger;
    
    /** * This class implements the consumer of the example * */
    public class Consumer implements Runnable { 
    
        /** * Buffer to save the events produced */
        private List<String> buffer;
    
        /** * Exchager to synchronize with the consumer */
        private final Exchanger<List<String>> exchanger;
    
        /** * Constructor of the class. Initializes its attributes * @param buffer Buffer to save the events produced * @param exchanger Exchanger to syncrhonize with the consumer */
        public Consumer(List<String> buffer, Exchanger<List<String>> exchanger){
            this.buffer=buffer;
            this.exchanger=exchanger;
        }
    
        /** * Main method of the producer. It consumes all the events produced by the Producer. After * processes ten events, it uses the exchanger object to synchronize with * the producer. It sends to the producer an empty buffer and receives a buffer with ten events */
        @Override
        public void run() {
            int cycle=1;
    
            for (int i=0; i<10; i++){
                System.out.printf("--Consumer: Cycle %d\n",cycle);
    
                try {
                    // Wait for the produced data and send the empty buffer to the producer
                    buffer=exchanger.exchange(buffer);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
                System.out.printf("----Consumer: %d\n",buffer.size());
    
                for (int j=0; j<10; j++){
                    String message=buffer.get(0);
                    System.out.printf("--Consumer: %s\n",message);
                    buffer.remove(0);
                }
    
                cycle++;
            }
    
        }
    
    }

结果

> –Consumer: Cycle 1  
> Producer: Cycle 1  
> Producer: Event 0  
> Producer: Event 1  
> Producer: Event 2  
> Producer: Event 3  
> Producer: Event 4  
> Producer: Event 5  
> Producer: Event 6  
> Producer: Event 7  
> Producer: Event 8  
> Producer: Event 9  
> Producer: 0  
> —-Consumer: 10  
> –Consumer: Event 0  
> –Consumer: Event 1  
> –Consumer: Event 2  
> Producer: Cycle 2  
> Producer: Event 10  
> –Consumer: Event 3  
> –Consumer: Event 4  
> –Consumer: Event 5  
> –Consumer: Event 6  
> Producer: Event 11  
> –Consumer: Event 7  
> –Consumer: Event 8  
> –Consumer: Event 9  
> –Consumer: Cycle 2  
> Producer: Event 12  
> Producer: Event 13  
> Producer: Event 14  
> Producer: Event 15  
> Producer: Event 16  
> Producer: Event 17  
> Producer: Event 18  
> Producer: Event 19  
> Producer: 0  
> Producer: Cycle 3  
> Producer: Event 20  
> Producer: Event 21  
> Producer: Event 22  
> Producer: Event 23  
> Producer: Event 24  
> Producer: Event 25  
> Producer: Event 26  
> —-Consumer: 10  
> –Consumer: Event 10  
> Producer: Event 27  
> –Consumer: Event 11  
> –Consumer: Event 12  
> Producer: Event 28  
> Producer: Event 29  
> –Consumer: Event 13  
> –Consumer: Event 14  
> –Consumer: Event 15  
> –Consumer: Event 16  
> –Consumer: Event 17  
> –Consumer: Event 18  
> –Consumer: Event 19  
> –Consumer: Cycle 3  
> —-Consumer: 10  
> –Consumer: Event 20  
> –Consumer: Event 21  
> –Consumer: Event 22  
> Producer: 0  
> Producer: Cycle 4  
> Producer: Event 30  
> –Consumer: Event 23  
> Producer: Event 31  
> Producer: Event 32  
> –Consumer: Event 24  
> Producer: Event 33  
> Producer: Event 34  
> Producer: Event 35  
> Producer: Event 36  
> Producer: Event 37  
> Producer: Event 38  
> Producer: Event 39  
> –Consumer: Event 25  
> –Consumer: Event 26  
> –Consumer: Event 27  
> –Consumer: Event 28  
> –Consumer: Event 29  
> –Consumer: Cycle 4  
> —-Consumer: 10  
> Producer: 0  
> Producer: Cycle 5  
> –Consumer: Event 30  
> –Consumer: Event 31  
> Producer: Event 40  
> –Consumer: Event 32  
> –Consumer: Event 33  
> –Consumer: Event 34  
> –Consumer: Event 35  
> –Consumer: Event 36  
> –Consumer: Event 37  
> –Consumer: Event 38  
> Producer: Event 41  
> –Consumer: Event 39  
> Producer: Event 42  
> –Consumer: Cycle 5  
> Producer: Event 43  
> Producer: Event 44  
> Producer: Event 45  
> Producer: Event 46  
> Producer: Event 47  
> Producer: Event 48  
> Producer: Event 49  
> Producer: 0  
> Producer: Cycle 6  
> —-Consumer: 10  
> Producer: Event 50  
> –Consumer: Event 40  
> Producer: Event 51  
> –Consumer: Event 41  
> Producer: Event 52  
> –Consumer: Event 42  
> Producer: Event 53  
> –Consumer: Event 43  
> Producer: Event 54  
> –Consumer: Event 44  
> Producer: Event 55  
> –Consumer: Event 45  
> Producer: Event 56  
> Producer: Event 57  
> Producer: Event 58  
> Producer: Event 59  
> –Consumer: Event 46  
> –Consumer: Event 47  
> –Consumer: Event 48  
> –Consumer: Event 49  
> –Consumer: Cycle 6  
> —-Consumer: 10  
> –Consumer: Event 50  
> –Consumer: Event 51  
> –Consumer: Event 52  
> –Consumer: Event 53  
> –Consumer: Event 54  
> –Consumer: Event 55  
> –Consumer: Event 56  
> –Consumer: Event 57  
> –Consumer: Event 58  
> –Consumer: Event 59  
> –Consumer: Cycle 7  
> Producer: 0  
> Producer: Cycle 7  
> Producer: Event 60  
> Producer: Event 61  
> Producer: Event 62  
> Producer: Event 63  
> Producer: Event 64  
> Producer: Event 65  
> Producer: Event 66  
> Producer: Event 67  
> Producer: Event 68  
> Producer: Event 69  
> Producer: 0  
> Producer: Cycle 8  
> —-Consumer: 10  
> –Consumer: Event 60  
> Producer: Event 70  
> –Consumer: Event 61  
> Producer: Event 71  
> Producer: Event 72  
> Producer: Event 73  
> Producer: Event 74  
> Producer: Event 75  
> –Consumer: Event 62  
> –Consumer: Event 63  
> –Consumer: Event 64  
> –Consumer: Event 65  
> –Consumer: Event 66  
> –Consumer: Event 67  
> –Consumer: Event 68  
> Producer: Event 76  
> –Consumer: Event 69  
> Producer: Event 77  
> –Consumer: Cycle 8  
> Producer: Event 78  
> Producer: Event 79  
> Producer: 0  
> Producer: Cycle 9  
> —-Consumer: 10  
> Producer: Event 80  
> –Consumer: Event 70  
> Producer: Event 81  
> –Consumer: Event 71  
> Producer: Event 82  
> –Consumer: Event 72  
> Producer: Event 83  
> –Consumer: Event 73  
> Producer: Event 84  
> –Consumer: Event 74  
> Producer: Event 85  
> –Consumer: Event 75  
> Producer: Event 86  
> –Consumer: Event 76  
> Producer: Event 87  
> –Consumer: Event 77  
> Producer: Event 88  
> –Consumer: Event 78  
> Producer: Event 89  
> –Consumer: Event 79  
> –Consumer: Cycle 9  
> —-Consumer: 10  
> –Consumer: Event 80  
> Producer: 0  
> Producer: Cycle 10  
> –Consumer: Event 81  
> Producer: Event 90  
> –Consumer: Event 82  
> Producer: Event 91  
> –Consumer: Event 83  
> Producer: Event 92  
> –Consumer: Event 84  
> Producer: Event 93  
> –Consumer: Event 85  
> Producer: Event 94  
> –Consumer: Event 86  
> Producer: Event 95  
> –Consumer: Event 87  
> Producer: Event 96  
> –Consumer: Event 88  
> Producer: Event 97  
> –Consumer: Event 89  
> Producer: Event 98  
> –Consumer: Cycle 10  
> Producer: Event 99  
> Producer: 0  
> —-Consumer: 10  
> –Consumer: Event 90  
> –Consumer: Event 91  
> –Consumer: Event 92  
> –Consumer: Event 93  
> –Consumer: Event 94  
> –Consumer: Event 95  
> –Consumer: Event 96  
> –Consumer: Event 97  
> –Consumer: Event 98  
> –Consumer: Event 99

## 工作原理 ##

消费者开始时是空白的buffer，然后调用Exchanger来与生产者同步。因为它需要数据来消耗。生产者也是从空白的buffer开始，然后创建10个字符串，保存到buffer，并使用exchanger与消费者同步。

在这儿，2个线程(生产者和消费者线程)都是在Exchanger里并交换了数据类型，所以当消费者从exchange() 方法返回时，它有10个字符串在buffer内。当生产者从 exchange() 方法返回时，它有空白的buffer来重新写入。这样的操作会重复10遍。

如你执行例子，你会发现生产者和消费者是如何并发的执行任务和在每个步骤它们是如何交换buffers的。与其他同步工具一样会发生这种情况，第一个调用 exchange()方法会进入休眠直到其他线程的达到。

### Exchanger 类有另外一个版本的exchange方法 ###

exchange(V data, long time, TimeUnit unit)：V是声明Phaser的参数种类(例子里是 List)。 此线程会休眠直到另一个线程到达并中断它，或者特定的时间过去了。TimeUnit类有多种常量：DAYS, HOURS, MICROSECONDS, MILLISECONDS, MINUTES, NANOSECONDS, 和 SECONDS。

转自： [ifeve][]

[ifeve]: http://ifeve.com/thread-synchronization-utilities-8/