防火墙设置

防火墙会屏蔽掉Kafka的访问，如果在内网限制很强的情况下，只能是逐个端口用 telnet 排查Kafka用到哪些端口，用lsof -i:<端口号>排查对应进程。

最粗暴的方式是关闭防火墙

# CentOS 6
service iptables stop
# CentOS 7
systemctl stop firewalld
systemctl disable firewalld

有可能要关闭SELINUX（不确定）

# 当前session生效
setenforce 0

要么编辑/etc/selinux/config，Session重新登录后生效

# SELINUX=enforcing # 默认值
SELINUX=disabled

文件资源句柄不足

当拿到机器未做操作系统层面优化时，往往会受到默认系统的参数限制，导致一些奇怪的异常。最常见的是文件资源句柄数量(nofile)和进程数量(nproc)的限制，可修改如下

# /etc/security/limits.conf文件
*    hard    nofile    65536
*    soft    nofile    65536
*    hard    nproc     65536
*    soft    nproc     65536

Session重新登录后生效，ulimit -a命令可以查看

Kafka listener设置

kafka的producer和consumer在查找broker时，依赖的主机名还是IP的情况比较迷，尤其是当hostname和请求中的Kafka broker配置不一致时，比较容易出问题，因此索性都改成IP比较不容易出问题

# 编辑 $KAFKA_HOME/config/server.properties文件
advertised.host.name=<Your IP>
advertised.port=9092
advertised.listeners=PLAINTEXT://<Your IP>:9092

kafka-python 的使用

kafka-python库用作python的依赖来访问kafka比较方便，注意在使用KafkaProducer时，可能会存在各种各样的坑，包括制定API版本、正确设定bootstrap_servers等。详细API使用说明可以参考官方KafkaProducer说明。

尤其注意参数 bootstrap_servers 使用broker的数组比较安全，尽量不要把所有broker信息都放到一个string里面

#!/usr/bin/env python
# encoding:utf-8
# 推荐的Producer声明方式
from kafka import KafkaProducerbrokers = ["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"]
producer = KafkaProducer(bootstrap_servers=brokers, acks=1, retries=3, max_block_ms=5000, api_version=(0, 8, 2))
"""acks = 1: Partition Leader做出回应max_block_ms = 5000: 如果不能正常拿到metadata, 5s就返回，默认是60sapi_version = (0, 8, 2): Kafka的版本元组，必须和Kafka Broker的版本一致，否则可能有问题，写大版本号(0, 10), (0, 8)之类的也可以
"""producer.send("topic_name", b"Hello World")
producer.flush()

Kafka 内存不稳定、吃CPU的优化

主要修改$KAFKA_HOME/bin/kafka-run-class.sh和$KAFKA_HOME/bin/kafka-server-start.sh

$KAFKA_HOME/bin/kafka-run-class.sh

1. G1GC垃圾回收能够避免掉CMS的一些问题
2. Kafka使用DirectMemory通信(基于NIO)，因此要设置大一些

KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:MaxDirectMemorySize=1024m -XX:+UseG1GC -Djava.awt.headless=true"

$KAFKA_HOME/bin/kafka-server-start.sh

1. -Xmn设置年轻代(Eden+S1+S2)大小，默认的年轻代尺寸过小，容易发生YoungGC
2. SurvivorRatio=Eden:S1，由于S1和S2大小一致，且主要用于GC时使用，所以可以评估年轻代大小，调整合适的比例。
3. 年轻代主要放在Eden区当中，Eden区大小=Xmn大小 * SurvivorRatio / (SurvivorRatio + 2)

export KAFKA_HEAP_OPTS="-Xmx8G -Xms2G -Xmn3G -XX:SurvivorRatio=6"

Kafka消息发送失败解决方案相关推荐

go 实现 kafka 消息发送、接收
引言网络上关于 go 实现 kafka 消息发送和接收的文章很多,但是实际操作起来又不是很清楚,本文在网络资源的基础上,结合自己搭建过程中遇到的问题进行了总结. 本文的实验主机:Mac笔记本. 一. ...
rabbitmq消息队列，消息发送失败，消息持久化，消费者处理失败相关
转:https://blog.csdn.net/u014373554/article/details/92686063 项目是使用springboot项目开发的,前是代码实现,后面有分析发送消息失败. ...
RocketMQ一行代码造成消息发送失败
这是我的第 198 期分享作者 | 丁威来源 | 中间件兴趣圈(ID:dingwpmz_zjj) 分享 | Java中文社群(ID:javacn666) 1.问题现象首先接到项目反馈使用 Roc ...
带你认识三种kafka消息发送模式
摘要:在kafka-0.8.2之后,producer不再区分同步(sync)和异步方式(async),所有的请求以异步方式发送,这样提升了客户端效率. 本文分享自华为云社区<kafka消息发送模 ...
RocketMQ 一行代码造成大量消息发送失败
作者 | 丁威来源 | 中间件兴趣圈问题现象首先接到项目反馈使用 RocketMQ 会出现如下错误: 错误信息关键点:MQBrokerException:CODE:2DESC:[TIMEOUT_ ...
soul服务器不稳定,soul聊天状态异常消息发送失败
大家好,我是时间财富网智能客服时间君,上述问题将由我为大家进行解答. soul聊天状态异常,消息发送失败的原因及解决方法如下: 1.发不出消息(显示红色感叹号),一般是由于网络不稳定,建议更换网络或者 ...
soul显示服务器异常,soul聊天状态异常消息发送失败
大家好,我是时间财富网智能客服时间君,上述问题将由我为大家进行解答. soul聊天状态异常,消息发送失败的原因及解决方法如下: 1.发不出消息(显示红色感叹号),一般是由于网络不稳定,建议更换网络或者 ...
极光im发送Message被拉黑和消息发送失败的处理办法
之前做的简单的聊天功能,今天让新加黑名单和免打扰功能,用的极光的im. 发现极光im的消息发送回调只有 code状态码和 String的提示语,如下: BasicCallback jmCallBac ...
Golang实现Kafka消息发送、接收
一:核心概念 kafka是消息中间件的一种,是一种分布式流平台,是用于构建实时数据管道和流应用程序.具有横向扩展,容错,wicked fast(变态快)等优点. kafka中涉及的名词: 消息记录(r ...

Kafka消息发送失败解决方案