/**
 * @Auther: baozi
 * @Date: 2019/6/4 11:01
 * @Description:
 */
abstract class MailSend {

    /**
     * 邮件缓存队列
     */
    private static BlockingQueue<Object> queue = new ArrayBlockingQueue<>(100);

    /**
     * 线程运行标志
     */
    private static AtomicBoolean run = new AtomicBoolean(false);

    /**
     * 发送邮件线程
     */
    private static Thread sendThread;

    /**
     * 开始发送线程
     * @return
     */
    public static boolean start() {
        if (!run.compareAndSet(false, true)) {
            return false;
        }

        sendThread = new Thread(() -> {

            while (run.get()) {
                try {
                    Object take = queue.take();
                    // todo do send mail
                    // MailUtils.sendMail(take); 线程安全? 
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }

        });
        sendThread.start();
        return true;
    }

    /**
     * 通知发送线程
     * @return
     */
    public static boolean stop() {
        if (!run.compareAndSet(true, false)) {
            return false;
        }
        sendThread.interrupt();
        return true;
    }

    /**
     * 发送邮件
     * @param object
     */
    public static void put(Object object) throws InterruptedException {
        queue.put(object);
    }

    public static void main(String[] args) throws InterruptedException {

        // 开始邮件发送者线程
        MailSend.start();

        // doing
        MailSend.put(new Object());
    }
}

public ThreadPoolExecutor(int corePoolSize, // 核心线程数
                              int maximumPoolSize, // 线程池允许最大的线程数
                              long keepAliveTime, // 空闲线程超时时间
                              TimeUnit unit, // 枚举时间单位
                              BlockingQueue<Runnable> workQueue) //阻塞队列

private final class Worker implements Runnable {
 
    final Thread thread;
    
    Runnable firstTask;
    
    Worker(Runnable firstTask) {
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    
    public void run() {
        runWorker(this);
    }
    
    final void runWorker(Worker w) {
        Runnable task = w.firstTask;
        w.firstTask = null;
        while (task != null || (task = getTask()) != null){
        task.run();
    }
        
    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
}

public class MessageCache {

    private static final Logger LOG = LoggerFactory.getLogger(Sender.class);

    protected BlockingQueue<ProducerBatch> queue = new ArrayBlockingQueue<>(10000);

    public void addMessage(Message message) {
        try {
            queue.put(meesage);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOG.error(e.getMessage(), e);
        }
    }
    /**
     * 获取缓存策略,能拿就拿,拿不到null就返回,最多拿的数量不大于maxSendRecords
     * 这里存在一种场景:如果消费者线程大于生产者线程速率,那么就永远不会拿到批量的数据.
     * 在实际场景中rocketmq的同步发送是比较慢的,所以不会出现上述情况
     * @return
     */
    public List<Message> getMessages() {
        List<Message> messages = new ArrayList<>(maxSendRecords);
        ProducerBatch producerBatch = queue.peek();
        // 上来就没消息,就休息1ms
        if (producerBatch == null) {
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                LOG.error(e.getMessage(), e);
            }
            return messages;
        }
        long size = 0;
        // 发送条件判断: 拿不到数据立即发 | 判断消息个数 | 判断总消息大小
        while (producerBatch != null && messages.size() < maxSendRecords ) {
            ProducerBatch batch = queue.poll();
            messages.add(batch.getMessage());
            size = producerBatch.getSize() + size;

            if(size > LIMIT_SIZE) {
                break;
            }

            producerBatch = queue.peek();
        }
        return messages;
    }
}

producer.send(record1);
producer.send(record2);

// 只有当所有参与复制的节点**全部都收到消息时，生产者才会收到一个来自服务器的成功响应
props.put(ProducerConfig.ACKS_CONFIG, "-1");
// 失败重试的次数(每次间隔100ms重试次数用完需要10多年)
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
//限制客户端在单个连接上能够发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
// 缓冲区已满或元数据不可用时的阻塞时间
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 1000);