Elasticsearch 5.4.3实战--Java API调用：批量写入数据

这个其实比较简单，直接上代码.

注意部分逻辑可以换成你自己的逻辑

package com.cs99lzzs.elasticsearch.service.imp;

import java.sql.Timestamp;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;


@Service("productIndexService")
public class ProductIndexServiceImp implements ProductIndexService {

    @Resource
    private TradeService               tradeService;

    @Resource
    private ProductService             productService;
    
    @Resource
    private ShopService                shopService;
    
    @Resource(name="esClient")
    Client esClient;

    @Value("${elasticsearch.index}")
    private String CLUSTER_INDEX;

    @Value("${elasticsearch.type}")
    private String CLUSTER_TYPE;
    

    private static final int           _DEFAULT_PAGE_SIZE = 50;

    private static final Logger        logger             = Logger.getLogger(ProductIndexServiceImp.class);

    @Override
    public void createIndex(Timestamp updateTime) {
        DecimalFormat decimalFormat = new DecimalFormat("#0.0");
        int page = 1;
        List<SKU> skus = null;
        
        long startTime = System.currentTimeMillis();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd  HH:mm:ss");
        logger.info("elasticsearch索引构建开始，开始时间：" + sdf.format(new Date()));
        
        while (true) {
            //获取当页的sku列表
            StringBuffer sb = new StringBuffer(1000);
            skus = productService.getSkus(updateTime, page, _DEFAULT_PAGE_SIZE);
            if(skus == null|| skus.isEmpty()){
                break;
            }
            //批量写入
            BulkRequestBuilder bulkRequest = esClient.prepareBulk();
            
            for(SKU sku:skus){
                try {
                    if(sku:skus == null){
                        logger.error("elasticsearch: skuId=" + sku.getId() + "对应的SPU或者Brand或者Cate对象为空，无需构建索引");
                        continue;
                    }
                    Map<String, Object> source = putToMap(sku);
                    bulkRequest.add(esClient
                            .prepareIndex(CLUSTER_INDEX, CLUSTER_TYPE, "" + elasticseachSku.getId())
                            .setSource(source));
                    sb.append(sku.getId()).append(",");
                } catch (Exception e) {
                    logger.error("更新elasticsearch索引出现异常, skuId=" + sku.getId() + ",exception info is:" 
                            + e.getMessage() + ", e.Cause is " + e.getCause());
                }
            } 
            
            BulkResponse response = bulkRequest.execute().actionGet();
            if (response == null || response.hasFailures()) {
                logger.error("elasticsearch 批量构建索引失败， failure message is: " + response.buildFailureMessage());
            } else {
                logger.info("elasticsearch 批量构建索引成功, skuId list is : " + sb.toString());
            }
            page ++;
        }
        logger.info("elasticsearch本次索引构建时间：" + (System.currentTimeMillis() - startTime)/1000 + "秒。");
        logger.info("elasticsearch索引构建任务结束，结束时间：" + sdf.format(new Date()));
    }

    /**
     * @param elasticseachSku
     * @return
     */
    private Map<String, Object> putToMap(Sku elasticseachSku) {
        
        Map<String, Object> source = new HashMap<String, Object>();
        source.put("brandZhName", elasticseachSku.getBrandZhName());
        source.put("brandEnName", elasticseachSku.getBrandEnName());
        source.put("brandAliases", elasticseachSku.getBrandAliases());
        source.put("aliases", elasticseachSku.getAliases());
        source.put("zhName", elasticseachSku.getZhName());
        source.put("enName", elasticseachSku.getEnName());
        source.put("brandZhName", elasticseachSku.getBrandZhName());
        
        /* suggester */
        List<String> nameList = new ArrayList<String>();
        if (StringUtils.isNotEmpty(elasticseachSku.getZhName())) {
            nameList.add(elasticseachSku.getZhName());
        }
        if (StringUtils.isNotEmpty(elasticseachSku.getBrandZhName())) {
            nameList.add(elasticseachSku.getBrandZhName());
        }
        if (StringUtils.isNotEmpty(elasticseachSku.getAliases())) {
            nameList.add(elasticseachSku.getAliases());
        }
        if (StringUtils.isNotEmpty(elasticseachSku.getEnName())) {
            nameList.add(elasticseachSku.getEnName());
        }
        if (StringUtils.isNotEmpty(elasticseachSku.getBrandEnName())) {
            nameList.add(elasticseachSku.getBrandEnName());
        }
        if (StringUtils.isNotEmpty(elasticseachSku.getBrandAliases())) {
            nameList.add(elasticseachSku.getBrandAliases());
        }
        if (nameList.size() > 0) {
            source.put("suggestName", nameList);
        }
        
        return source;
    }
}