讲个故事

面试官：IPv4地址可以转为Long类型的数字知道吧？你写一下这个转换的代码！

对计算机基础逐渐模糊的小徐一脸懵逼，毕竟工作中很少会用到，只记得IP地址和整数是可以相互转换的，但是从来没有自己实现过。于是在大脑中飞速计算。过了一会，思路出现了：IP地址分为四段，每段都是0~255之间的数，每段可以用8位来装下它，4×8=32位，也就是可以将IP地址转为32位的整数。咦？面试官居然让转成Long，但是Long有64位啊！一个int就搞定了，为什么要转为Long呢？但是自己没实现过，心里没底，再一犯嘀咕，然后回答不出来。面试没通过！！！

之后小徐一直心心念念这个问题，回来立马自己写了代码实现了一下，果然是int就搞定了，之前的思路一点都没错！！！

面试有时考验的不只是技术，还有自信心！！！

 

IP地址是一个32位的二进制数，通常被分割为4个“8位二进制数”（也就是4个字节）。IP地址通常用“点分十进制”表示成（a.b.c.d）的形式，其中，a,b,c,d都是0~255之间的十进制整数。

例：点分十进IP地址（100.4.5.6），实际上是32位二进制数（01100100.00000100.00000101.00000110）。

 

为什么8位能存储0~255之间的数字？

一个字节8个位，每个位就只有0跟1两种情况，8个位能表示2的8次方即256，范围0-255（带负值的话范围在：-128~127）；

8位最小二进制为00000000转十进制即为0

8位最大二进制为11111111转十进制即为255

所以8位可存储0~255之间的数

IP地址本身就是一个32位的二进制数，只是通常被以a.b.c.d的形式表示而已。

那么为什么要将IP转为数字呢？

其实就是时间换空间的一种方式。

String类型的IP占用7个字节到15个字节0.0.0.0~255.255.255.255，而int只需要4个字节。

IP字符串转换为Int

/**
 * 将 ip 字符串转换为 int 类型的数字
 * <p>
 * 思路就是将 ip 的每一段数字转为 8 位二进制数，并将它们放在结果的适当位置上
 *
 * @param ipString ip字符串，如 127.0.0.1
 * @return ip字符串对应的 int 值
 */
public static int ip2Int(String ipString) {
    // 取 ip 的各段
    String[] ipSlices = ipString.split("\\.");
    int rs = 0;
    for (int i = 0; i < ipSlices.length; i++) {
        // 将 ip 的每一段解析为 int，并根据位置左移 8 位
        int intSlice = Integer.parseInt(ipSlices[i]) << 8 * i;
        // 或运算
        rs = rs | intSlice;
    }
    return rs;
}

有一个技巧，就是 或运算。就是将每段的 int 值左移到恰当的位置后跟保存结果的 int 值进行或运算。

以 255.255.255.255 这个地址为例，上面的或运算过程如下：

00000000 00000000 00000000 00000000
00000000 00000000 00000000 11111111
------------或运算------------
00000000 00000000 00000000 11111111
00000000 00000000 11111111 00000000
------------或运算------------
00000000 00000000 11111111 11111111
00000000 11111111 00000000 00000000
------------或运算------------
00000000 11111111 11111111 11111111
11111111 00000000 00000000 00000000
-----------最终结果------------
11111111 11111111 11111111 11111111

那么如何将 int 再转为字符串的表示法呢？

Int转换为IP字符串

思路是一样的，将 int 值的 32 位分为 4 个 8 位数字，然后这 4 个 8 位的数字用 0~255 的数字进行表示，用点号分隔即可。我们也基于位运算，7行代码即可实现：

/**
 * 将 int 转换为 ip 字符串
 *
 * @param ipInt 用 int 表示的 ip 值
 * @return ip字符串，如 127.0.0.1
 */
public static String int2Ip(int ipInt) {
    String[] ipString = new String[4];
    for (int i = 0; i < 4; i++) {
        // 每 8 位为一段，这里取当前要处理的最高位的位置
        int pos = i * 8;
        // 取当前处理的 ip 段的值
        int and = ipInt & (255 << pos);
        // 将当前 ip 段转换为 0 ~ 255 的数字，注意这里必须使用无符号右移
        ipString[i] = String.valueOf(and >>> pos);
    }
    return String.join(".", ipString);
}

这里使用与运算来取每次处理的 ip 片段。取最高的 8 位时，涉及到符号的处理，因此在将每段 8 位转为 0~255 的数字时必须使用无符号右移运算，否则最后处理的部分因为符号问题会不准确。

测试一下

public static void main(String[] args) {
    String[] ips4Test = new String[]{"0.0.0.0", "127.0.0.1", 
    	"192.168.1.1", "255.0.0.255", "255.255.255.255"};
    for (String ip : ips4Test) {
        test(ip);
    }
}

public static void test(String ip) {
    int ipInt = ip2Int(ip);
    String ipString = int2Ip(ipInt);
    System.out.println("用于测试的ip地址: " + ip + ", int表示: " + ipInt + ", 二进制: " + Long.toBinaryString(ipInt)
            + ", 转回String: " + ipString + "，与测试 ip 地址是否相等: " + ip.equals(ipString));
}

打印结果：

用于测试的ip地址: 0.0.0.0, int表示: 0, 二进制: 0, 转回String: 0.0.0.0，与测试 ip 地址是否相等: true
用于测试的ip地址: 127.0.0.1, int表示: 16777343, 二进制: 1000000000000000001111111, 转回String: 127.0.0.1，与测试 ip 地址是否相等: true
用于测试的ip地址: 192.168.1.1, int表示: 16885952, 二进制: 1000000011010100011000000, 转回String: 192.168.1.1，与测试 ip 地址是否相等: true
用于测试的ip地址: 255.0.0.255, int表示: -16776961, 二进制: 1111111111111111111111111111111111111111000000000000000011111111, 转回String: 255.0.0.255，与测试 ip 地址是否相等: true
用于测试的ip地址: 255.255.255.255, int表示: -1, 二进制: 1111111111111111111111111111111111111111111111111111111111111111, 转回String: 255.255.255.255，与测试 ip 地址是否相等: true

注意：这里相互转换的算法是配套的，不同的转换算法计算的 int 值可能会不一样，因为虽然都是处理 ip 的 4 个部分，但是它们的结合顺序可以不一样，因此以怎样的顺序搭配转为 int，就应该以相同的顺序解析为 String。

回顾前面做一些实验，你会发现，实现一的并发性能高于实现二、三。暂且不关心BlockingQueue的具体实现，来分析看如何优化实现三（与实现二的思路相同，性能相当）的性能。

分析实现三的瓶颈

最好的查证方法是记录方法执行时间，这样可以直接定位到真正的瓶颈。但此问题较简单，我们直接用“瞪眼法”分析。

实现三的并发瓶颈很明显，因为在锁 BUFFER_LOCK 看来，任何消费者线程与生产者线程都是一样的。换句话说，同一时刻，最多只允许有一个线程（生产者或消费者，二选一）操作缓冲区 buffer。

而实际上，如果缓冲区是一个队列的话，“生产者将产品入队”与“消费者将产品出队”两个操作之间没有同步关系，可以在队首出队的同时，在队尾入队。理想性能可提升至实现三的两倍。

去掉这个瓶颈

那么思路就简单了：需要两个锁 CONSUME_LOCK与PRODUCE_LOCK，CONSUME_LOCK控制消费者线程并发出队，PRODUCE_LOCK控制生产者线程并发入队；相应需要两个条件变量NOT_EMPTY与NOT_FULL，NOT_EMPTY负责控制消费者线程的状态（阻塞、运行），NOT_FULL负责控制生产者线程的状态（阻塞、运行）。以此让优化消费者与消费者（或生产者与生产者）之间是串行的；消费者与生产者之间是并行的。

package com.github.xuchengen.concurrent.impl;

import com.github.xuchengen.concurrent.AbsConsumer;
import com.github.xuchengen.concurrent.AbsProducer;
import com.github.xuchengen.concurrent.Model;
import com.github.xuchengen.concurrent.Task;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 基于锁条件实现
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/12
 */
public class LockConditionModel2 implements Model {

    private final Lock CONSUME_LOCK = new ReentrantLock();

    private final Condition NOT_EMPTY = CONSUME_LOCK.newCondition();

    private final Lock PRODUCE_LOCK = new ReentrantLock();

    private final Condition NOT_FULL = PRODUCE_LOCK.newCondition();

    private final Buffer<Task> buffer = new Buffer<>();

    private AtomicInteger bufLen = new AtomicInteger(0);

    private final int cap;

    private final AtomicInteger increTaskNo = new AtomicInteger(0);

    public LockConditionModel2(int cap) {
        this.cap = cap;
    }

    @Override
    public Runnable newRunnableConsumer() {
        return new ConsumerImpl();
    }

    @Override
    public Runnable newRunnableProducer() {
        return new ProducerImpl();
    }

    private class ConsumerImpl extends AbsConsumer {
        @Override
        public void consume() throws InterruptedException {
            int newBufSize = -1;
            CONSUME_LOCK.lockInterruptibly();
            try {
                while (bufLen.get() == 0) {
                    System.out.println("buffer is empty...");
                    NOT_EMPTY.await();
                }

                Task task = buffer.poll();

                assert task != null;
                // 固定时间范围的消费，模拟相对稳定的服务器处理过程
                Thread.sleep(500 + (long) (Math.random() * 500));
                System.out.println("consume: " + task.no);

                newBufSize = bufLen.decrementAndGet();
                if (newBufSize > 0) {
                    // 非空唤醒继续消费
                    NOT_EMPTY.signalAll();
                }
            } finally {
                CONSUME_LOCK.unlock();
            }
            if (newBufSize < cap) {
                PRODUCE_LOCK.lockInterruptibly();
                try {
                    // 未满唤醒生产者继续生产
                    NOT_FULL.signalAll();
                } finally {
                    PRODUCE_LOCK.unlock();
                }
            }
        }
    }


    private class ProducerImpl extends AbsProducer {
        @Override
        public void produce() throws InterruptedException {
            // 不定期生产，模拟随机的用户请求
            Thread.sleep((long) (Math.random() * 1000));
            int newBufSize = -1;
            PRODUCE_LOCK.lockInterruptibly();
            try {
                while (bufLen.get() == cap) {
                    System.out.println("buffer is full...");
                    NOT_FULL.await();
                }
                Task task = new Task(increTaskNo.getAndIncrement());
                buffer.offer(task);
                newBufSize = bufLen.incrementAndGet();
                System.out.println("produce: " + task.no);
                if (newBufSize < cap) {
                    // 未满唤醒继续消费
                    NOT_FULL.signalAll();
                }
            } finally {
                PRODUCE_LOCK.unlock();
            }
            if (newBufSize > 0) {
                CONSUME_LOCK.lockInterruptibly();
                try {
                    // 非空唤醒消费者继续消费
                    NOT_EMPTY.signalAll();
                } finally {
                    CONSUME_LOCK.unlock();
                }
            }
        }
    }

    private static class Buffer<E> {
        private Node head;
        private Node tail;

        Buffer() {
            // dummy node
            head = tail = new Node(null);
        }

        public void offer(E e) {
            tail.next = new Node(e);
            tail = tail.next;
        }

        public E poll() {
            head = head.next;
            E e = head.item;
            head.item = null;
            return e;
        }

        private class Node {
            E item;
            Node next;

            Node(E item) {
                this.item = item;
            }
        }
    }

    public static void main(String[] args) {
        Model model = new LockConditionModel2(3);
        for (int i = 0; i < 2; i++) {
            new Thread(model.newRunnableConsumer()).start();
        }
        for (int i = 0; i < 5; i++) {
            new Thread(model.newRunnableProducer()).start();
        }
    }

}

需要注意的是，由于需要同时在UnThreadSafe的缓冲区 buffer 上进行消费与生产，我们不能使用实现二、三中使用的队列了，需要自己实现一个简单的缓冲区 Buffer。Buffer要满足以下条件：

• 在头部出队，尾部入队
• 在poll()方法中只操作head
• 在offer()方法中只操作tail

还能进一步优化吗

我们已经优化掉了消费者与生产者之间的瓶颈，还能进一步优化吗？

如果可以，必然是继续优化消费者与消费者（或生产者与生产者）之间的并发性能。然而，消费者与消费者之间必须是串行的，因此，并发模型上已经没有地方可以继续优化了。

仔细分析下，实现四中的signalAll都可以换成signal。这里为了屏蔽复杂性，回避了这个优化。

不过在具体的业务场景中，一般还能够继续优化。如：

• 并发规模中等，可考虑使用CAS代替重入锁
• 模型上不能优化，但一个消费行为或许可以进一步拆解、优化，从而降低消费的延迟
• 一个队列的并发性能达到了极限，可采用“多个队列”（如分布式消息队列等）

这4种写法的本质相同——都是在使用或实现BlockingQueue。实现一直接使用BlockingQueue，实现四实现了简单的BlockingQueue，而实现二、三则实现了退化版的BlockingQueue（性能降低一半）。

我们要保证理解wait && notify机制。实现时可以使用Object类提供的wait()方法与notifyAll()方法，但更推荐的方式是使用java.util.concurrent包提供的Lock && Condition。

package com.github.xuchengen.concurrent.impl;

import com.github.xuchengen.concurrent.AbsConsumer;
import com.github.xuchengen.concurrent.AbsProducer;
import com.github.xuchengen.concurrent.Model;
import com.github.xuchengen.concurrent.Task;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 基于锁条件实现
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/12
 */
public class LockConditionModel1 implements Model {

    private final Lock BUFFER_LOCK = new ReentrantLock();

    private final Condition BUFFER_COND = BUFFER_LOCK.newCondition();

    private final Queue<Task> buffer = new LinkedList<>();

    private final int cap;

    private final AtomicInteger increTaskNo = new AtomicInteger(0);

    public LockConditionModel1(int cap) {
        this.cap = cap;
    }

    @Override
    public Runnable newRunnableConsumer() {
        return new ConsumerImpl();
    }

    @Override
    public Runnable newRunnableProducer() {
        return new ProducerImpl();
    }

    private class ConsumerImpl extends AbsConsumer {
        @Override
        public void consume() throws InterruptedException {
            BUFFER_LOCK.lockInterruptibly();
            try {
                while (buffer.size() == 0) {
                    // 当缓冲中元素为空则阻塞该线程与此同时会唤醒生产者线程进行生产
                    BUFFER_COND.await();
                }
                Task task = buffer.poll();
                assert task != null;
                // 固定时间范围的消费，模拟相对稳定的服务器处理过程
                Thread.sleep(500 + (long) (Math.random() * 500));
                System.out.println("consume: " + task.no);
                // 消费后唤醒生产者线程进行生产
                BUFFER_COND.signalAll();
            } finally {
                BUFFER_LOCK.unlock();
            }
        }
    }

    private class ProducerImpl extends AbsProducer {
        @Override
        public void produce() throws InterruptedException {
            // 不定期生产，模拟随机的用户请求
            Thread.sleep((long) (Math.random() * 1000));
            BUFFER_LOCK.lockInterruptibly();
            try {
                while (buffer.size() == cap) {
                    // 当缓冲中元素已满则阻塞该线程与此同时会唤醒消费者线程进行消费
                    BUFFER_COND.await();
                }
                Task task = new Task(increTaskNo.getAndIncrement());
                buffer.offer(task);
                System.out.println("produce: " + task.no);
                // 生产后唤醒消费者线程进行消费
                BUFFER_COND.signalAll();
            } finally {
                BUFFER_LOCK.unlock();
            }
        }
    }

    public static void main(String[] args) {
        Model model = new LockConditionModel1(3);
        for (int i = 0; i < 2; i++) {
            new Thread(model.newRunnableConsumer()).start();
        }
        for (int i = 0; i < 5; i++) {
            new Thread(model.newRunnableProducer()).start();
        }
    }

}

该写法的思路与实现二的思路完全相同，仅仅将锁与条件变量换成了Lock和Condition。

如果不能将并发与容量控制都封装在缓冲区中，就只能由消费者与生产者完成。最简单的方案是使用朴素的wait && notify机制。

package com.github.xuchengen.concurrent.impl;

import com.github.xuchengen.concurrent.AbsConsumer;
import com.github.xuchengen.concurrent.AbsProducer;
import com.github.xuchengen.concurrent.Model;
import com.github.xuchengen.concurrent.Task;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 基于对象锁wait和notify实现
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public class WaitNotifyModel implements Model {

    private final Object BUFFER_LOCK = new Object();

    private final Queue<Task> buffer = new LinkedList<>();

    private final int cap;

    private final AtomicInteger increTaskNo = new AtomicInteger(0);

    public WaitNotifyModel(int cap) {
        this.cap = cap;
    }

    @Override
    public Runnable newRunnableConsumer() {
        return new ConsumerImpl();
    }

    @Override
    public Runnable newRunnableProducer() {
        return new ProducerImpl();
    }

    private class ConsumerImpl extends AbsConsumer {
        @Override
        public void consume() throws InterruptedException {
            synchronized (BUFFER_LOCK) {
                while (buffer.size() == 0) {
                    // 当缓冲中元素为空则阻塞该线程与此同时会唤醒生产者线程进行生产
                    BUFFER_LOCK.wait();
                }
                Task task = buffer.poll();
                assert task != null;
                // 固定时间范围的消费，模拟相对稳定的服务器处理过程
                Thread.sleep(500 + (long) (Math.random() * 500));
                System.out.println("consume: " + task.no);
                // 消费后唤醒生产者线程进行生产
                BUFFER_LOCK.notifyAll();
            }
        }
    }

    private class ProducerImpl extends AbsProducer {
        @Override
        public void produce() throws InterruptedException {
            // 不定期生产，模拟随机的用户请求
            Thread.sleep((long) (Math.random() * 1000));
            synchronized (BUFFER_LOCK) {
                while (buffer.size() == cap) {
                    // 当缓冲中元素已满则阻塞该线程与此同时会唤醒消费者线程进行消费
                    BUFFER_LOCK.wait();
                }
                Task task = new Task(increTaskNo.getAndIncrement());
                buffer.offer(task);
                System.out.println("produce: " + task.no);
                // 生产后唤醒消费者线程进行消费
                BUFFER_LOCK.notifyAll();
            }
        }
    }

    public static void main(String[] args) {
        Model model = new WaitNotifyModel(3);
        for (int i = 0; i < 2; i++) {
            new Thread(model.newRunnableConsumer()).start();
        }
        for (int i = 0; i < 5; i++) {
            new Thread(model.newRunnableProducer()).start();
        }
    }

}

BlockingQueue的写法最简单。核心思想是，把并发和容量控制封装在缓冲区中。而BlockingQueue的性质天生满足这个要求。

package com.github.xuchengen.concurrent.impl;

import com.github.xuchengen.concurrent.AbsConsumer;
import com.github.xuchengen.concurrent.AbsProducer;
import com.github.xuchengen.concurrent.Model;
import com.github.xuchengen.concurrent.Task;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 基于阻塞队列的缓冲实现
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public class BlockingQueueModel implements Model {

    private final BlockingQueue<Task> queue;

    private final AtomicInteger increTaskNo = new AtomicInteger();

    public BlockingQueueModel(int cap) {
        this.queue = new LinkedBlockingQueue<>(cap);
    }

    @Override
    public Runnable newRunnableConsumer() {
        return new ConsumerImpl();
    }

    @Override
    public Runnable newRunnableProducer() {
        return new ProducerImpl();
    }

    private class ConsumerImpl extends AbsConsumer {

        @Override
        public void consume() throws InterruptedException {
            Task task = queue.take();
            // 固定时间范围的消费，模拟相对稳定的服务器处理过程
            Thread.sleep(500 + (long) (Math.random() * 500));
            System.out.println("consume: " + task.no);
        }
    }

    private class ProducerImpl extends AbsProducer {

        @Override
        public void produce() throws InterruptedException {
            // 不定期生产，模拟随机的用户请求
            Thread.sleep((long) (Math.random() * 1000));
            Task task = new Task(increTaskNo.getAndIncrement());
            System.out.println("produce: " + task.no);
            queue.put(task);
        }
    }

    public static void main(String[] args) {

        Model model = new BlockingQueueModel(3);

        //2个消费者
        for (int i = 0; i < 2; i++) {
            new Thread(model.newRunnableConsumer()).start();
        }

        //5个生产者
        for (int i = 0; i < 5; i++) {
            new Thread(model.newRunnableProducer()).start();
        }
    }

}

由于操作“出队/入队+日志输出”不是原子的，所以上述日志的绝对顺序与实际的出队/入队顺序有出入，但对于同一个任务号task.no，其consume日志一定出现在其produce日志之后，即：同一任务的消费行为一定发生在生产行为之后。缓冲区的容量留给读者验证。符合两个验证条件。

BlockingQueue写法的核心只有两行代码，并发和容量控制都封装在了BlockingQueue中，正确性由BlockingQueue保证。面试中首选该写法，自然美观简单。

考查Java的并发编程时，手写“生产者-消费者模型”是一个经典问题。有如下几个考点：

• 对Java并发模型的理解
• 对Java并发编程接口的熟练程度
• bug free
• coding style

JDK版本：oracle java 1.8.0_102

本文主要归纳了4种写法，阅读后，最好在白板上练习几遍，检查自己是否掌握。这4种写法或者编程接口不同，或者并发粒度不同，但本质是相同的——都是在使用或实现BlockingQueue。

生产者-消费者模型

网上有很多生产者-消费者模型的定义和实现。本文研究最常用的有界生产者-消费者模型，简单概括如下：

• 生产者持续生产，直到缓冲区满，阻塞；缓冲区不满后，继续生产
• 消费者持续消费，直到缓冲区空，阻塞；缓冲区不空后，继续消费
• 生产者可以有多个，消费者也可以有多个

可通过如下条件验证模型实现的正确性：

• 同一产品的消费行为一定发生在生产行为之后
• 任意时刻，缓冲区大小不小于0，不大于限制容量

该模型的应用和变种非常多，不赘述。

准备

面试时可语言说明以下准备代码。关键部分需要实现，如AbsConsumer。

下面会涉及多种生产者-消费者模型的实现，可以先抽象出关键的接口，并实现一些抽象类：

package com.github.xuchengen.concurrent;

/**
 * 生产者接口
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public interface Producer {

    /**
     * 生产者负责生产
     *
     * @throws InterruptedException 线程意外终止异常
     */
    void produce() throws InterruptedException;

}

package com.github.xuchengen.concurrent;

/**
 * 消费者接口
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public interface Consumer {

    /**
     * 消费者负责消费
     *
     * @throws InterruptedException 线程意外终止异常
     */
    void consume() throws InterruptedException;

}

package com.github.xuchengen.concurrent;

/**
 * 抽象生产者
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public abstract class AbsProducer implements Producer, Runnable {

    @Override
    public void run() {
        while (true) {
            try {
                produce();
            } catch (InterruptedException e) {
                e.printStackTrace();
                break;
            }
        }
    }
}

package com.github.xuchengen.concurrent;

/**
 * 抽象生产者
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public abstract class AbsConsumer implements Consumer, Runnable {

    @Override
    public void run() {
        while (true) {
            try {
                consume();
            } catch (InterruptedException e) {
                e.printStackTrace();
                break;
            }
        }
    }

}

不同的模型实现中，生产者、消费者的具体实现也不同，所以需要为模型定义抽象工厂方法：

package com.github.xuchengen.concurrent;

/**
 * 模型
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public interface Model {

    /**
     * 实例化一个消费者
     *
     * @return Runnable
     */
    Runnable newRunnableConsumer();

    /**
     * 实例化一个生产者
     *
     * @return Runnable
     */
    Runnable newRunnableProducer();

}

我们将Task作为生产和消费的单位：

package com.github.xuchengen.concurrent;

/**
 * 任务
 * 作者：徐承恩
 * 邮箱：xuchengen@gmail.com
 * 日期：2019/12/11
 */
public class Task {

    /**
     * 任务号
     */
    public int no;

    public Task(int no) {
        this.no = no;
    }
}

如果需求还不明确（这符合大部分工程工作的实际情况），建议边实现边抽象，不要“面向未来编程”。