兄弟们兄弟们。我又来了。

之前Java程序直接写入ES。 不知道有时候网络不好。还是啥情况。

我这个写入就超时了。 然后这条消息就会丢失。 系统也嗷嗷报错。针对于这个情况。

修改了一下 写入ES的方式。

程序将消息放入-> Kafka -> logstash进行消费后输出至 -> ES中。

好了。废话不多说。老样子。大象装冰箱一共分3步

步骤1、Java程序集成kafka-》用的springBoot

maven-加入jar包

    <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
    </dependency>

spring配置文件。增加配置

spring:
  kafka:
    bootstrap-servers: kafka地址
    producer:
      # 发生错误后，消息重发的次数。
      retries: 0
      #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
      batch-size: 16384
      # 设置生产者内存缓冲区的大小。
      buffer-memory: 33554432
      # 键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
      # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
      # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
      acks: 1

增加kafka发送处理类

@Service
@Slf4j
public class KafkaServiceImpl implements KafkaService {

  @Autowired
  private KafkaTemplate<String, Object> kafkaTemplate;

  public static final String TOPIC_STR = "pile-order-charge-process";

  @Override
  public void send (OrderChargeProcessEs orderChargeProcessEs) {
    String obj2String = JSONObject.toJSONString(orderChargeProcessEs);
    log.info("准备发送消息为：{}", obj2String);
    //发送消息
    ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_STR, obj2String);
    future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
      @Override
      public void onFailure (Throwable throwable) {
        //发送失败的处理
        log.error(TOPIC_STR + " - 生产者 发送消息失败：" + throwable.getMessage());
      }

      @Override
      public void onSuccess (SendResult<String, Object> stringObjectSendResult) {
        //成功的处理
        //log.info(TOPIC_STR + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
      }
    });
  }
}

步骤2、修改logstash管道配置、修改配置文件

管道的配置文件

input {
  kafka{
    bootstrap_servers => "kafka地址"
    topics_pattern  => "pile-order-charge-process.*"
    consumer_threads => 12
    decorate_events => true
    codec => "json"
    auto_offset_reset => "latest"
    group_id => "pile-order-charge-process"
  }
}
filter {
}
output {
  elasticsearch {
    hosts => ["es集群地址]
    user => "es账号"
    password => "es密码"
    index => "pile-order-charge-process-%{+YYYY.MM}"
  }
}

logstash的logstash.yml配置文件-将对应的管道ID设置进来。 对应上面创建的管道ID

xpack.management.pipeline.id: ["pile-order-charge-process"]

步骤三、重启logstash、kafka观察主体、kibana上观察logstash消费情况

通过kafka-manager看到队列已经创建出来了

 

通过kibana可以看到。管道已经创建。并有输入输出的数据了

本期内容到此为止了。兄弟们。 如果觉得有用 。真的帮助到你了。 麻烦兄弟给我点个赞