最近要做一个初始版的数据仓库项目，有一大批的MYSQL基础数据要通过整合后写入到ES，有部分数据量非常庞大，单线程批量写入会耗时很久，就想到了使用线程池来多线程做写入操作。

这个写法不要局限于我这个应用场景，别的应用场景也可以修改一下里面的逻辑。

java版本要求在 1.8以上

附代码：

import org.apache.commons.collections.CollectionUtils;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 批量多线程执行写入任务
 */
public class BatchThreadSave {

    /**
     * 批量多线程执行写入任务
     * @param action 返回的截取数据表  自行写操作方法
     * @param dataList 总数据表
     * @param threadCount 需要的线程总数  0 < threadCount < 11
     * @param <F> 指定数据表泛型
     */
    public static <F> void save(Executor<? super List<F>> action, List<F> dataList, int threadCount) {
        if (CollectionUtils.isEmpty(dataList)) {
            return;
        }
        if (threadCount < 1 || threadCount > 10) {
            return;
        }
        int threadSize = dataList.size() / (threadCount - 1);
        // 总数据条数
        int dataSize = dataList.size();
        // 分段数
        int threadNum = dataSize / threadSize + 1;
        // 创建一个线程池
        ExecutorService exec = Executors.newFixedThreadPool(threadCount);
        // 定义一个任务集合
        List<F> cutList = null;
        // 确定每条线程的数据
        try {
            for (int i = 0; i < threadNum; i++) {
                if (i == threadNum - 1) {
                    cutList = dataList.subList(threadSize * i, dataSize);
                } else {
                    cutList = dataList.subList(threadSize * i, threadSize * (i + 1));
                }
                List<F> finalCutList = cutList;
                exec.execute(() -> {
                    action.runAction(finalCutList);
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭线程池
            exec.shutdown();
        }
    }

    public interface Executor<T extends List> {
        void runAction(T t);
    }




}

那该怎么用呢？

public void buildUserIndex() {
        List<User> users = userMapper.findAll();
        log.info("共获取到用户：" + users.size());

        BatchThreadSave.save(finalCutList -> {
            int j = 0;
            for (User user : finalCutList) {

                j++;
                log.info(Thread.currentThread().getName() + "   当前执行到" + user.getId() + "  剩余：" + (finalCutList.size() - j));

                //finalCutList 这里就是被截取后的数据， 只管写后面的实现
                //这里执行你的业务逻辑就行了  比如保存  数据处理等等
                
                
            }
        }, users, 10);

    }

我有一个应用，需要使用pandas去mysql数据库里用pd. read_sql方法读取数据，进行一定处理后通过数据可视化组件Pyecharts在网页上进行可视化展示，图形为折线图，在显示时发现有些系列（series)的值不显示，查看pyecharts网页端的JavaScript脚本代码，发现Series的值都显示为null，后来查了资料，发现是pandas的数值类型有些跟pyecharts不兼容，pyecharts官网是这么说的：

Note: 在使用 Pandas&Numpy 时，请确保将数值类型转换为 python 原生的 int/float。比如整数类型请确保为 int，而不是 numpy.int32

在网上查了很多资料，比如直接使用df.astype(‘int’)对整个数据表进行转换，或者编写自定义函数然后调用df.apply()方法，都是报错，因为我的pandas原始数据里面有int,object等不同的数据类型，不能直接转换。后来就弄了一个for循环，循环每1行，然后对每一行使用.apply()方法调用lambda函数进行转换，数据就正常显示，虽然用循环感觉比较笨，但还是开心，解决了问题。毕竟现在电脑配置那么高，不太需要考虑程序效率，能把问题解决就行了。

核心代码如下：

#从数据库读取数据
df=pd.read_sql(sql=sql,con=conn,index_col="日期",coerce_float=False)
#折线图
line = Line(init_opts=opts.InitOpts(width="1200px",height = "600px"))
line.add_xaxis(df.columns.tolist())        #一定要加tolist()方法，不然会没有任何内容。

#设置Y轴反转，X轴朝下，不用对其零值
yaxis_opts=opts.AxisOpts(is_inverse= True,is_show=True,axisline_opts=opts.AxisLineOpts(is_on_zero=False)))

for i in range(0,df.shape[0],1):            #轮循df里的所有行，添加到折线图。
  series_name = df.iloc[i].name   #系列名称
  y_axis_value=df.iloc[i].apply(lambda x: x.astype('float') if x is not None else x)    #这里把所有非空值强行转换为float
  line.add_yaxis(series_name=series_name, y_axis=y_axis_value,label_opts=opts.LabelOpts(is_show=True),
        linestyle_opts=opts.LineStyleOpts(width=2),)
        #添加Y轴数据

【亲测有效】C# 使用lambda表达式修改集合的某个属性、字段的值

前言：我们通常在开发的时候会面对各种需求，而我们又喜欢写代码的时候可以简洁、高雅、高效率。lambda表达式的出现正好能够帮助我们减少了一定的代码量，提高写代码的效率。

正文：
我们经常需要对一个集合做一些自定义的修改，而C#又不允许使用foreach或者for修改集合内容，那么下面我介绍三种方式实现修改集合的某个属性、字段的值，用起来简洁方便，亲测有效。

下面有实体类SelectOption

public class SelectOption
{
    public string Title
    {
    	get;set;
    }
    public object Value
    {
    	get;set;
    }
}

实现的关键代码：

//实例化泛型集合
List<SelectOption> selectOptions = new List<SelectOption>();
//加入测试数据1
selectOptions.Add(new SelectOption()
{
    Title = "标题1",
    Value = "001"
});
//加入测试数据2
selectOptions.Add(new SelectOption()
{
    Title = "标题2",
    Value = "002"
});

//第一种方式，使用Select
selectOptions = selectOptions.Select(c =>
{
    c.Title = "测试：" + c.Title;
    return c;
}).ToList();

//第二种方式，使用ForEach
selectOptions.ForEach(item =>
{
    item.Title = "测试：" + item.Title;
});

//第三种方式，使用FindAll
selectOptions = selectOptions.FindAll(delegate(SelectOption obj) {
    obj.Title = "测试：" + obj.Title;
    return obj != null;
});

总结：三种实现方式都是可以在方法体里面加入逻辑判断代码，看个人喜好使用哪一种方式都行。

写完收工。