// 大体流程
1. 从原始借据表bp_collect_loan_src同步数据到bp_collect_loan，同步批次号为batch_no的数据。(同步数据时，每个线程处理数量默认为10000，若是当前批次数据量超过10000，会采用多线程来执行)
2. 针对每一个原始借据，数据库里面只有可能有一条与之对应的借据，所以将bp_collect_loan存入数据库的操作采用的是insertOrUpdate操作(任务执行失败后，不会对数据库中的数据造成影响)。
3. 借据同步完成之后，需要执行归案任务，就是将多个借据加工成为一个案件，以人为维度进行催收。
4. 归案完成之后，需要跑规则引擎来执行分案，通过规则引擎分案之后，会在数据库表bp_collect_divide表里面生成分案记录。
5. 将这些分案记录针对案件进行分案(分案记录里面存储了两类，针对个人的分案，针对组的分案)

// 使用到的技术点(分布式锁 redis队列)
1. 由于任务是部署在多台机器上的，多个进程执行任务时，需要获取分布式锁。项目中采用redis来实现了分布式锁，上锁命令采用set命令，外加是否存在，过期时间参数来实现。而释放锁的命令采用lua脚本来实现，因为会有两步骤操作，首先式判断上锁客户端和释放锁的客户端是否一致（存储在value中），其次是删除该锁对应的key，lua脚本保证了该操作的原子性。
2. 在执行归案的过程中，针对每一个案件，需要将该案件分出去，会过规则引擎。而将分案请求交给规则引擎有两种方式，第一种是先将该案件构建成为请求，然后将该请求放到redis的同步队列里面，规则引擎那边从同步队列里面来进行消费。第二种是若同步队列已满，或者放入到同步队列出错，则会将请求直接交给规则引擎来处理。

// 技术难点-如何克服

// 创建线程池的工具类Executors
public class Executors {
    // 创建一个只有一个工作线程的线程池
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    
    // 创建一个固定数目线程的线程池(核心线程数和最大线程数相同)
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    // 创建一个cache线程池，核心线程大小为0
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    
    
    // 线程工厂，负责为线程池中的Worker创建线程-》可以创建个性化的线程
    static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            // 返回的线程又将Worker做了一层封装,当该线程执行strart方法时,将会调用Worker的run方法
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }    
}


// jdk定义的线程池
public class ThreadPoolExecutor extends AbstractExecutorService {
    // 核心线程数
    private volatile int corePoolSize;
    // 最大线程数
    private volatile int maximumPoolSize;
    // 阻塞队列
    private final BlockingQueue<Runnable> workQueue;
    // 非工作线程存活时间
    private volatile long keepAliveTime;
    // 创建线程的工厂
    private volatile ThreadFactory threadFactory;
    // 线程池的拒绝策略
    private volatile RejectedExecutionHandler handler;
    // 存储创建好的Worker线程
    private final HashSet<Worker> workers = new HashSet<Worker>();
    // 主锁,对workers进行操作时会上锁
    private final ReentrantLock mainLock = new ReentrantLock();
    
    // 定义线程池内部的工作线程
    private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
		// 当前Worker持有的线程
        final Thread thread;
        // 当前Worker执行的第一个任务
        Runnable firstTask;
        // 记录当前Worker已经完成的任务数量
        volatile long completedTasks;

        Worker(Runnable firstTask) {
            // 将aqs中的state值由0设置为-1,禁止中断(上锁了)
            setState(-1); 
            this.firstTask = firstTask;
            // 调用工厂为当前的Worker创建一个Thread—>传入的是当前this对象
            this.thread = getThreadFactory().newThread(this);
        }
		
        // 当前Worker线程启动需要执行的方法，该方法会由内部属性thread调用start方法时触发。
        public void run() {
            runWorker(this);
        }

		// 判断线程是否被独占
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }
		
        // 尝试获取锁
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }
   	
    // Worker线程执行的真正逻辑
    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock();//执行任务之前允许被打断
        boolean completedAbruptly = true;
        try {
            // 当前worker的task不为null或者是阻塞队列不为null,worker线程会一直运行
            while (task != null || (task = getTask()) != null) {
                // 上锁
                w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        // 调用worker的run方法
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
    
    
    
    
    
    
    
    // 初始化一个线程池
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
    
    // 1.若是当前工作线程数小于核心线程数, 会去创建一个核心线程。
    // 2.若是当前工作线程数大于等于核心线程数，将任务放入到阻塞队列。
    // 3.若是阻塞队列已满，会去创建非核心线程。
    // 4.若是创建非核心线程也失败，执行拒绝策略。
    public void execute(Runnable command) {
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }
    
    // 1.尝试着将工作线程数加1，workers的个数。
    // 2.创建一个Worker线程，并将其加入到workers数组中。
    // 3.若是加入到workers数组成功，调用worker的thread的start方法，启动线程。
    // 4.线程启动之后，会调用worker的run方法，而run方法又是调用runWorker(this)方法来执行的。
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        // 自旋
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();
                if (runStateOf(c) != rs)
                    continue retry;
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive())
                            throw new IllegalThreadStateException();
                        // 将创建好的Worker加入到workers数组中
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

    
    
    
    
    
    
    
    
}

// spring线程池->ThreadPoolTaskExecutor
public class ThreadPoolTaskExecutor {
    private final Object poolSizeMonitor = new Object();
    // 核心线程数
    private int corePoolSize = 1;
    // 最大线程数
    private int maxPoolSize = 2147483647;
    // 非核心线程空闲最大存活时间
    private int keepAliveSeconds = 60;
    // 缓存队列容量
    private int queueCapacity = 2147483647;
    // 是否允许核心线程池超时等待(设置为false,当到达一定时间没有任务,线程池会自动关闭)
    private boolean allowCoreThreadTimeOut = false;
    
    private TaskDecorator taskDecorator;
    private ThreadPoolExecutor threadPoolExecutor;

    public ThreadPoolTaskExecutor() {
        
    }
    
// 因为spring线程池提供了更加灵活的配置，项目中一般使用的是spring的线程池。

/**
 * NIO Server服务
 */
class NIOServer{
    // Selector->注册channel
    Selector selector = null;
    // ServerSocketChannel->服务端channel,类似ServerSocket
    ServerSocketChannel serverSocketChannel;
    // 处理selector轮询事件
    private ChannelHandle handle;
    // NIO服务关闭标识
    private volatile boolean stop = false;

    NIOServer(int port) {
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            // 设置channel为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            // 为channel绑定端口
            serverSocketChannel.bind(new InetSocketAddress(port));
            // 将channel注册到selector上，监听连接事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            handle = new ChannelHandle();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动服务
     * @throws IOException
     */
    public void start() throws IOException {
        while (!stop) {
            // 获取到等待处理的IO事件数量
            int readyChannels = selector.select();
            // 若是等待处理的IO事件数量为0,不处理
            if (readyChannels == 0) {
                continue;
            }
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            // 处理这些SelectionKey
            while (iterator.hasNext()) {
                // 获取到该key
                SelectionKey key = iterator.next();
                // 移除该key
                iterator.remove();
                // 分别处理各自事件
                if (key.isAcceptable()) {
                    handle.handleAccept(key);
                } else if (key.isReadable()) {
                    handle.handleRead(key);
                } else if (key.isWritable()) {
                    handle.handleWrite(key);
                }
            }
        }
    }

	// 停止该服务
    public void stop() {
        this.stop = true;
    }
}


/**
 * 针对selector上不同事件的处理类
 */
class ChannelHandle {
    /**
     * 处理连接事件
     * @param key
     * @throws IOException
     */
    public void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        SocketChannel sc = ssc.accept();
        sc.configureBlocking(false);
        sc.register(key.selector(), SelectionKey.OP_READ);
    }

    /**
     * 处理可读事件
     * @param key
     * @throws IOException
     */
    public void handleRead(SelectionKey key) throws IOException {
        SocketChannel sc = (SocketChannel) key.channel();
        // fixme 对读取到的数据进行处理->相关协议解析
        sc.register(key.selector(), SelectionKey.OP_WRITE);
    }

    /**
     * 处理可写事件
     * @param key
     * @throws IOException
     */
    public void handleWrite(SelectionKey key) throws IOException {
        System.out.println("处理写数据");
        // fixme 对输出结果按照相关协议进行封装
        String header = "HTTP/1.1 200 OK\r\n";
        StringBuffer result = new StringBuffer(header);
        result.append("Content-Type:application/json\n");
        result.append("\r\n");
        result.append("hello,world");
        SocketChannel channel = (SocketChannel)key.channel();
        ByteBuffer wrap = ByteBuffer.wrap(result.toString().getBytes());
        channel.write(wrap);
        channel.close();
    }
}

// 调用类
public class NIOUtil {
    public static void main(String[] args) {
        // 启动一个基于NIO的服务
        NIOServer nioServer = new NIOServer(8070);
        try {
            nioServer.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

public class IOUtil {

    public static void main(String[] args) {
        IOServer.createServer();
    }
}


class IOServer {
    
    private static volatile boolean stop = false;
    /**
     * 开启服务
     */
    static void createServer() {
        ExecutorService executorService =
                Executors.newFixedThreadPool(3);
        ServerSocket serverSocket = null;
        try {
            serverSocket = new ServerSocket(8090);
            System.out.println("服务器在端口8090上启动。。。");
            while (!stop) {
                try {
                    Socket socket = serverSocket.accept();
                    // 将任务提交给线程池来处理
                    executorService.execute(new SocketHandle(socket));
                } catch (Exception e) {
                    e.printStackTrace();
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (serverSocket !=  null) {
                    serverSocket.close();
                }
                executorService.shutdown();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 关闭服务
     */
    static void stopServer() {
        stop = true;
    }
}


/**
 * socket处理线程类
 */
class SocketHandle implements Runnable {
    private Socket socket;

    SocketHandle(Socket socket) {
        super();
        this.socket = socket;
    }

    @Override
    public void run() {
        // 处理输入
        InputStream is = null;
        InputStreamReader isr = null;
        BufferedReader br = null;
        try {
            is = socket.getInputStream();
            isr = new InputStreamReader(is);
            br = new BufferedReader(isr);
            String s = null;
            while ((s = br.readLine()) != null && s.length() > 0) {
                System.out.println(s);
                s = null;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {

        }
        // 处理输出 \r\n 回车换行
        OutputStream os = null;
        try {
            os = socket.getOutputStream();
            os.write("HTTP/1.1 200 OK\r\n".getBytes());
            os.write("Content-Type:application/json\n".getBytes());
            os.write("\r\n".getBytes());
            os.write("hello,world".getBytes());
            os.flush();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}