专栏首页杂烩kafka连接异常 原

kafka连接异常 原

kafka版本是0.10.2.1 本地java客户端版本是0.8.1.1

主要两个错误

第一个是连接拒绝

kafka Connection refused: no further information

然后发现第二个错误

Selector.poll(Selector.java:276) - Error in I/O with localhost/127.0.0.1

怀疑是ip绑定有问题,编辑server.properties,指定ip地址

advertised.host.name=ip地址

重启后,运行客户端,抛出另外一个问题

KafkaException: Failed to construct kafka producer

可能是因为客户端版本过低导致,升级客户端版本后,运行正常。

以下测试demo

生产者

public class ProducerDemo {

	private static final Logger LOG = Logger.getLogger(ProducerDemo.class);

	private static Properties properties = new Properties();

	static {
		properties.put("bootstrap.servers", "172.16.41.151:9092");
		properties.put("acks", "all");
		properties.put("retries", 0);
		properties.put("batch.size", 16384);
		properties.put("linger.ms", 1);
		properties.put("buffer.memory", 33554432);
		properties.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
		properties.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
	}

	public void produce() {
		KafkaProducer<byte[], byte[]> kafkaProducer = new KafkaProducer<byte[], byte[]>(properties);
		ProducerRecord<byte[], byte[]> kafkaRecord = new ProducerRecord<byte[], byte[]>("canal-t_canal",
				"kkk".getBytes(), "ssss".getBytes());
		kafkaProducer.send(kafkaRecord, new Callback() {
			public void onCompletion(RecordMetadata metadata, Exception e) {
				if (null != e) {
					LOG.info("the offset of the send record is {" + metadata.offset() + "}");
					LOG.error(e.getMessage(), e);
				}
				LOG.info("complete!");
			}
		});
		kafkaProducer.close();
	}

	public static void main(String[] args) {
		ProducerDemo kafkaProducerTest = new ProducerDemo();
		for (int i = 0; i < 10; i++) {
			kafkaProducerTest.produce();
		}
	}

}

消费者

public class ConsumerDemo {

	private static final String topic = "canal-t_canal";

	private static final Integer threads = 3;

	public static void main(String[] args) {

		Properties props = new Properties();

		props.put("zookeeper.connect", "172.16.41.151:2181");

		props.put("group.id", "1111");

		props.put("zookeeper.session.timeout.ms", "40000");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.commit.interval.ms", "1000");

		ConsumerConfig config = new ConsumerConfig(props);

		ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);

		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();

		topicCountMap.put(topic, 1);

		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);

		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get("canal-t_canal");

		for (final KafkaStream<byte[], byte[]> kafkaStream : streams) {

			new Thread(new Runnable() {

				public void run() {

					for (MessageAndMetadata<byte[], byte[]> mm : kafkaStream) {

						String msg = new String(mm.message());

						System.out.println(msg);

					}

				}

			}).start();

		}

	}

}

pom

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>scc-stu</groupId>
	<artifactId>stu-kafka</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>war</packaging>
	<name>stu-kafka</name>
	<description>stu-kafka</description>
	<dependencies>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka-clients</artifactId>
			<version>0.10.0.0</version>
		</dependency>

		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka_2.11</artifactId>
			<version>0.10.0.0</version>
		</dependency>
		<!-- log4j -->
		<dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.6.1</version>
        </dependency>
	</dependencies>
</project>

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Invalid character found in the request target. The valid characters are defined in RFC 7230 and RF

    原因是升级后的tomcat设置了url参数解析规则,限制非法字符,如  |  ,解决办法是:

    尚浩宇
  • Docker容器之最小JDK基础镜像 原

        2、因为java依赖glibc,所以基础镜像使用的是alpine-glibc而非alpine,alpine-glibc大概是11.1M。

    尚浩宇
  • websocket 原

         WebSocket protocol 是HTML5一种新的协议。它实现了浏览器与服务器全双工通信(full-duplex)。   

    尚浩宇
  • (9)Shark工作流的实现和WMFC&OMG规范的对比

    <?xml:namespace prefix = o ns = "urn:schemas-microsoft-com:office:office" />

    田春峰-JCJC错别字检测
  • [silverlight基础]仿文字连接跑马灯效果-高手绕道

    运行效果如下: ? 分析示意图: ? 代码:  <Canvas x:Name="a" Background="AliceBlue" MouseEnter="a_...

    菩提树下的杨过
  • 【Rust每周一库】Yew - Rust语言实现的WebAssembly多线程前端框架

    Yew是一个设计先进的Rust前端框架,目的是使用WebAssembly来创建多线程前端web应用。

    MikeLoveRust
  • Kotlin中,函数作为参数,T.()->Unit 和 ()->Unit 的区别

    Xiaolei123
  • dubbo源码解析——cluster

    将 Directory 中的多个 Invoker 伪装成一个 Invoker(伪装过程用到loadBalance),对上层透明,伪装过程包含了容错逻辑,调用失败...

    用户5325874
  • 互联网分层架构,为啥要前后端分离?

    画外音:“别人在搞xxoo技术”一定不能成为,一家公司推动“xxoo技术”的理由。

    前端黑板报
  • 人工智能教授警告:陪伴机器人对老人很危险

    Margaret Boden教授。图片来源:JAY WILLIAMS 一名人工智能教授发出这样的警告:老人家的机器人陪伴“在情感上很危险”,永远无法代替真实人类...

    AI科技评论

扫码关注云+社区

领取腾讯云代金券