systemctl stop firewalld.service

docker pull wurstmeister/kafka
docker pull wurstmeister/zookeeper

docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

docker run --name kafka01 \
-p 9092:9092 \
-e KAFKA_BROKER_ID=0 \
-e KAFKA_ZOOKEEPER_CONNECT=192.168.2.115:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.2.115:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
-d  wurstmeister/kafka  

​docker exec -it kafka01 /bin/bash

/opt/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.2.115:2181 --replication-factor 1 --partitions 1 --topic my_log

/opt/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.2.115:2181

public class KafkaSender<T> {

	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;

	private final static String MY_TOPTIC_LOG = "goods_mylog";

	/**
	 * kafka 发送消息
	 *
	 * @param obj
	 *            消息对象
	 */
	public void send(T obj) {
		String jsonObj = JSON.toJSONString(obj);
		log.info("------------ message = {}", jsonObj);

		// 发送消息
		ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(MY_TOPTIC_LOG, jsonObj);
		future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
			@Override
			public void onFailure(Throwable throwable) {
				log.info("Produce: The message failed to be sent:" + throwable.getMessage());
			}

			@Override
			public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
				// TODO 业务处理
				log.info("Produce: The message was sent successfully:");
				log.info("Produce: _+_+_+_+_+_+_+ result: " + stringObjectSendResult.toString());
			}
		});
	}
}

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import com.alibaba.fastjson.JSONObject;
import com.pitch.base.BaseApiService;
import com.pitch.base.BaseResponse;
import com.pitch.goods.ProductSearchService;
import com.pitch.member.output.dto.ProductDto;
import com.pitch.product.es.entity.ProductEntity;
import com.pitch.product.es.reposiory.ProductReposiory;
import com.pitch.product.kafka.KafkaSender;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.querydsl.QPageRequest;
import org.springframework.web.bind.annotation.RestController;
import ma.glasnost.orika.MapperFactory;
import ma.glasnost.orika.impl.DefaultMapperFactory;

@RestController
public class ProductSearchServiceImpl extends BaseApiService<List<ProductDto>> implements ProductSearchService {
	@Autowired
	private ProductReposiory productReposiory;
	@Autowired
	private KafkaSender<JSONObject> kafkaSender;
	private int i = 0;

	@Override
	public BaseResponse<List<ProductDto>> search(String name) {
		// 1.拼接查询条件
		BoolQueryBuilder builder = QueryBuilders.boolQuery();
		// 2.模糊查询name\subtitle\detail字段
		builder.must(QueryBuilders.multiMatchQuery(name, "name", "subtitle", "detail"));
		Pageable pageable = new QPageRequest(0, 5);
		// 3.调用ES接口查询
		Page<ProductEntity> page = productReposiory.search(builder, pageable);
		// 4.获取集合数据
		List<ProductEntity> content = page.getContent();
		// 5.将entity转换dto
		MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
		List<ProductDto> mapAsList = mapperFactory.getMapperFacade().mapAsList(content, ProductDto.class);
		i++;
		if (i%3==0){
			try {
				System.out.println(i/0);
			} catch (Exception e) {
				JSONObject errorJson = new JSONObject();
				JSONObject logJson = new JSONObject();
				SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置日期格式
				logJson.put("request_time", df.format(new Date()));
				logJson.put("error_info", e);
				errorJson.put("request_error", logJson);
				kafkaSender.send(errorJson);
			}

		}

		return setResultSuccess(mapAsList);
	}
}

input {
  kafka {
    bootstrap_servers => "192.168.2.115:9092"
    topics => ["goods_mylog"]
  }
}
output {
    stdout { codec => rubydebug }
    elasticsearch {
       hosts => ["192.168.2.115:9200","192.168.2.117:9200"]
       index => "goods_mylog"
    }
}

// 启动elasticsearch
cd /usr/local/elasticsearch6.4/bin
./elasticsearch -d
// 启动kibana
su root
cd /usr/local/kibana6.4/bin
nohup ./kibana &
// 启动logstash  等待几分钟
cd /usr/local/logstash-6.4.3
./bin/logstash -f goods_mylog.conf

GET goods_mylog/doc/_search
{
  "query":{
      "match":{
        "message": "request_error"
      }   
  }
}