java连接ElasticSearch集群

你的名字 2022-06-16 08:57 283阅读 0赞

/*
     *es配置类
     *
     */
    
    @Configuration
    public class ElasticSearchDataSourceConfigurer {
    
        private static final Logger LOG = LogManager.getLogger(ElasticSearchDataSourceConfigurer.class);
    
    
        @Bean
        public TransportClient getESClient() {
            //设置集群名称
            Settings settings = Settings.builder().put("cluster.name", "bigData-cluster").put("client.transport.sniff", true).build();
            //创建client
            TransportClient client = null;
            try {
                client = new PreBuiltTransportClient(settings)
                        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(""), 9300));//集群ip
                LOG.info("ESClient连接建立成功");
            } catch (UnknownHostException e) {
                LOG.info("ESClient连接建立失败");
                e.printStackTrace();
            }
            return client;
        }
    
    
    } /**
     * Simple to Introduction
     *
     * @Description: [添加类]
     */
    @Repository
    public class UserDaoImpl implements userDao {
    
    	private static final String INDEXNAME = "user";//小写
    	private static final String TYPENAME = "info";
    
    	@Resource
    	TransportClient transportClient;
    
    	@Override
    	public int addUser(User[] user) {
    		IndexResponse indexResponse = null;
    		int successNum = 0;
    		for (int i = 0; i < user.length; i++) {
    			UUID uuid = UUID.randomUUID();
    			String str = uuid.toString();
    			String jsonValue = null;
    			try {
    				jsonValue = JsonUtil.object2JsonString(user[i]);
    				if (jsonValue != null) {
    					indexResponse = transportClient.prepareIndex(INDEXNAME, TYPENAME, str).setSource(jsonValue)
    							.execute().actionGet();
    					successNum++;
    				}
    			} catch (JsonProcessingException e) {
    				e.printStackTrace();
    			}
    
    		}
    		return successNum;
    	}
    
    } /**
     *批量插入
     */
    public static void bathAddUser(TransportClient client, List<User> users) {
    
    		BulkRequestBuilder bulkRequest = transportClient.prepareBulk();
    		for (int i = 0; i < users.size(); i++) {
    			UUID uuid = UUID.randomUUID();
    			String str = uuid.toString();
    
    			String jsonValue = null;
    			try {
    				jsonValue = JsonUtil.object2JsonString(users.get(i));
    			} catch (JsonProcessingException e) {
    				e.printStackTrace();
    			}
    			bulkRequest.add(client.prepareIndex("user", "info", str).setSource(jsonValue));
    			// 一万条插入一次
    			if (i % 10000 == 0) {
    				bulkRequest.execute().actionGet();
    			}
    			System.out.println("已经插入第" + i + "多少条");
    		}
    
    	}