from queue import Queue
import threading
import time

from openpyxl import load_workbook

class test_q(object):
    def __init__(self):
        self.q1 = Queue()
        self.q2 = Queue()
        self.q3 = Queue()

    # 队列一种放入数据
    def put_data_1(self):
        i = 0
        while i < 100000:
            self.q1.put(str(i))
            i += 1

    # 从队列q1中取出数据，并放入队里q2
    def worker1(self):
        while True:
            num1 = self.q1.get()
            if num1:
                print('从q1中拿到的数据是11111111111%s' % num1)
                self.q2.put('放入q2数据%s' % num1)
            self.q1.task_done()
    # 从队列q2中取出数据放入q3
    def worker2(self):
        while True:
            re_num = self.q2.get()
            if re_num:
                print('从q2中拿到数据是22222222222222%s' % re_num)
                self.q3.put('放入3的数据%s' % re_num)
            self.q2.task_done()

    #  从队列q3中取出数据写入excel表格
    def worker3(self, table, loc):
        while True:
            loc.acquire()
            res_3 = self.q3.get()
            if res_3:
                print('从q3中取出的数据333333******%s' % res_3)
                # with open('G:\\vw\zmyw_test.txt', 'a+', encoding='utf8') as f:
                #     f.write(res_3+'\n')
                table.append([res_3])
            self.q3.task_done()
            loc.release()

    def run(self):
        # 线程列表
        thread_list = []

        # 队列q1放入数据的线程
        q1_t = threading.Thread(target=self.put_data_1)
        thread_list.append(q1_t)

        # 从队列q1取出数据放入q2的线程
        q2_t = threading.Thread(target=self.worker1)
        thread_list.append(q2_t)

        # 从队列q2中取出数据放入q3的线程
        q3_t = threading.Thread(target=self.worker2)
        thread_list.append(q3_t)

        # 打开excel表格
        open_excel = load_workbook('G:\\vw\zmyw_test.xlsx')
        table = open_excel['Sheet1']

        #  从队列q3中取出数据写入excel表格
        loc = threading.Lock()
        for i in range(5):
            w_excel = threading.Thread(target=self.worker3, args=(table,loc))
            thread_list.append(w_excel)
        for i in thread_list:
            print('开启线程')
            i.setDaemon(True)
            i.start()

        for j in [self.q1, self.q2, self.q3]:
            print('_____')
            j.join()
        print('***************')
        open_excel.save('G:\\vw\zmyw_test.xlsx')

if __name__=='__main__':
    te = test_q()
    te.run()

大数据中，多线程分段list操作， demo如下

/**
 * 每500条数据开启一条线程
 *
 * @Author: wpf
 * @Date: 16:05 2018/5/21
 * @Description: 
 * @param  * @param null  
 * @return   
 */
private void threadExec2(List<SysCompany> list) throws ExecutionException, InterruptedException {
    // 开始时间
    long start = System.currentTimeMillis();

    LOG.info("list的大小：" + list.size());

    // 每500条数据开启一条线程
    int threadSize = 500;
    // 总数据条数
    int dataSize = list.size();
    // 线程数
    int threadNum = dataSize / threadSize + 1;
    // 定义标记,过滤threadNum为整数
    boolean special = dataSize % threadSize == 0;
    // 创建一个线程池
    ExecutorService exec = Executors.newFixedThreadPool(threadNum);
    // 定义一个任务集合
    List<Callable<SysCompany>> tasks = new ArrayList<Callable<SysCompany>>();
    Callable<SysCompany> task = null;
    List<SysCompany> cutList = null;
    // 确定每条线程的数据
    for (int i = 0; i < threadNum; i++) {
        if (i == threadNum - 1) {
            if (special) {
                break;
            }
            cutList = list.subList(threadSize * i, dataSize);
        } else {
            cutList = list.subList(threadSize * i, threadSize * (i + 1));
        }
        // LOG.info("第" + (i + 1) + "组：" + cutList.toString());
        final List<SysCompany> listStr = cutList;
        task = new Callable<SysCompany>() {

            @Override
            public SysCompany call() throws Exception {
                for(SysCompany c : listStr){
                    if(StringUtils.isNotEmpty(c.getUsername())){
                        LOG.info("timer：在执行中..." + c.getUsername());
                        // 业务...
                        sysCompanyService.getComCodeByInfoCompanyUser(c.getUsername());
                    }
                }
                return null;
            }
        };
        // 这里提交的任务容器列表和返回的Future列表存在顺序对应的关系
        tasks.add(task);
    }
    List<Future<SysCompany>> results = exec.invokeAll(tasks);
    /*for (Future<SysCompany> future : results) {
        SysCompany c = future.get();
    }*/
    // 关闭线程池
    exec.shutdown();
    LOG.info("线程任务执行结束");
    LOG.info("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
}

/**
 * 固定开辟8线程
 *
 * @Author: wpf
 * @Date: 16:05 2018/5/21
 * @Description:
 * @param  * @param null
 * @return
 */
private void threadExec1(List<SysCompany> list) throws ExecutionException, InterruptedException{
    // 开始时间
    long start = System.currentTimeMillis();

    // 总数据条数
    int listSize = list.size();
    LOG.info("总记录数为："+listSize);
    // 线程数
    int threadNum = 8;

    // 创建一个线程池
    ExecutorService exec = Executors.newFixedThreadPool(threadNum);
    // 定义一个任务集合
    List<Callable<Integer>> tasks = new ArrayList<Callable<Integer>>();
    Callable<Integer> task = null;
    List<SysCompany> cutList = null;

    // 均分算法
    Map<Integer,Long> map = allotOfAverage1(threadNum,listSize);
    int endValue = 0;
    int startValue = 0;
    int i = 0;
    for (Long v : map.values()) {
        endValue = (int)(startValue+v);
        cutList = list.subList(startValue, endValue);
        startValue += v;
        LOG.info("第" + ++i +"组大小为："+cutList.size());
        final List<SysCompany> listStr = cutList;
        task = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                for(SysCompany c : listStr){
                    if(StringUtils.isNotEmpty(c.getUsername())){

                        // LOG.info("timer：在执行中..." + c.getUsername());
                        // 业务...
                        sysCompanyService.getComCodeByInfoCompanyUser(c.getUsername());
                    }
                }
                return 1;
            }
        };
        // 这里提交的任务容器列表和返回的Future列表存在顺序对应的关系
        tasks.add(task);
    }

    List<Future<Integer>> results = exec.invokeAll(tasks);
    for (Future<Integer> future : results) {
        LOG.info(future.get());
    }
    // 关闭线程池
    exec.shutdown();
    LOG.info("线程任务执行结束");
    LOG.info("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
}

/*
 * 平均分配
 */
public static Map<Integer,Long> allotOfAverage1(int threadNum, long listSize){
    Map<Integer,Long> allot=new HashMap<Integer,Long>(); //保存分配的信息
    for(int i=0;i<listSize;i++){
        int j=i%threadNum;
        if(allot.containsKey(j)){
            allot.put(j, allot.get(j) +1);
        }else{
            allot.put(j, 1L);
        }
    }
    return allot;
}