Kafka的发送与接收数据

2020-02-02 19:04:13 阅读：965 来源： 互联网

标签：parseObject System kafka 发送 println import 接收数据 com Kafka

首先配置端口号还有主题应
其次，在spring-bean.xml里面加载配置的生产者与消费者
如下：
consumer:消费者的配置

<?xml version="1.0" encoding="UTF-8"?>

<bean id="consumerProperties" class="java.util.HashMap">
	<constructor-arg>
		<map>
			<!--Kafka服务地址 -->
			<entry key="bootstrap.servers" value="192.168.27.128:9092" />
			<!--Consumer的组ID，相同group.id的consumer属于同一个组。一个组中的不同成员,相同的消息只能有一个人收到 -->
			<entry key="group.id" value="test-consumer-group" />
			<!--如果此值设置为true，consumer会周期性的把当前消费的offset值保存到zookeeper。当consumer失败重启之后将会使用此值作为新开始消费的值。 -->
			<entry key="enable.auto.commit" value="true" />
			<!--网络请求的socket超时时间。实际超时时间由max.fetch.wait + socket.timeout.ms 确定 -->
			<entry key="session.timeout.ms" value="15000 " />

			<entry key="key.deserializer"
				value="org.apache.kafka.common.serialization.StringDeserializer" />

			<entry key="value.deserializer"
				value="org.apache.kafka.common.serialization.StringDeserializer" />
		</map>
	</constructor-arg>
</bean>


<!-- 创建consumerFactory bean -->
<bean id="consumerFactory"
	class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
	<constructor-arg>
		<ref bean="consumerProperties" />
	</constructor-arg>
</bean>


<bean id="messageListenerContainer"
	class="org.springframework.kafka.listener.KafkaMessageListenerContainer"
	init-method="doStart">
	<constructor-arg ref="consumerFactory" />
	<constructor-arg ref="containerProperties" />
</bean>



<!-- 记得修改主题 -->
<bean id="containerProperties" class="org.springframework.kafka.listener.ContainerProperties">
	<!-- 构造函数 就是 主题的参数值 -->
	<constructor-arg value="1708E" />
	<!-- 要在此处声明自己定义的监听器的类 -->
	<property name="messageListener" ref="msgListener" />
</bean>



<!--指定具体监听类的bean 这个类需要我们手工写-->
<bean id="msgListener" class="com.lixuecheng.kafka.MentListener" />

producer:生产者的配置 <?xml version="1.0" encoding="UTF-8"?>

    <!--参数配置 -->
<bean id="producerProperties" class="java.util.HashMap">
	<constructor-arg>
		<map>
			<!-- kafka服务地址，可能是集群 value="localhost:9092,localhost:9093,localhost:9094"-->
			<entry key="bootstrap.servers" value="192.168.27.128:9092" />
			
			<!-- 有可能导致broker接收到重复的消息-->
			<entry key="retries" value="0" />
			<!-- 每次批量发送消息的数量 -->
			<entry key="batch.size" value="1638" />
			<!-- 默认0ms，在异步IO线程被触发后（任何一个topic，partition满都可以触发） -->
			<entry key="linger.ms" value="1" />
			
			<!--producer可以用来缓存数据的内存大小。如果数据产生速度大于向broker发送的速度，producer会阻塞或者抛出异常 -->
			<entry key="buffer.memory" value="33554432 " />
			
			<entry key="key.serializer"
				value="org.apache.kafka.common.serialization.StringSerializer" />
				
			<entry key="value.serializer"
				value="org.apache.kafka.common.serialization.StringSerializer" />
		</map>
	</constructor-arg>
</bean>

<!-- 创建kafkatemplate需要使用的producerfactory bean -->
<bean id="producerFactory"
	class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
	<constructor-arg>
		<ref bean="producerProperties" />
	</constructor-arg>
</bean>

<!-- 创建kafkatemplate bean，使用的时候，只需要注入这个bean，即可使用template的send消息方法 -->
<bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
	<constructor-arg ref="producerFactory" />
	<!--设置对应topic 如果主题不存在,会自动创建-->
	<property name="defaultTopic" value="1708E" />
</bean>

然后测试类加载spring-beans.xml文件
在代码中运行测试
要创建一个监听kafka的配置
package com.lixuecheng.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.MessageListener;

import com.alibaba.fastjson.JSON;
import com.lixuecheng.entity.Investment;
import com.lixuecheng.service.InMentService;

public class MentListener implements MessageListener<String, String> {

@Autowired
private InMentService inMentService;

@Override
public void onMessage(ConsumerRecord<String, String> data) {

	String value = data.value();
	System.err.println("11111" + value);
	System.out.println("111111111111111111111");
	try {
		Thread.sleep(500);
	} catch (InterruptedException e) { // TODO Auto-generated catch block
		e.printStackTrace();
	}
	System.out.println("收到了！！！！");

	Investment parseObject = JSON.parseObject(value, Investment.class);

	System.out.println(parseObject);

	 inMentService.add(parseObject);

}

}
最后要在虚拟机中启动kafka及其相关应用即可

wuhui_cheng 发布了3 篇原创文章 · 获赞 0 · 访问量 38 私信关注

标签：parseObject,System,kafka,发送,println,import,接收数据,com,Kafka
来源： https://blog.csdn.net/wuhui_cheng/article/details/104147849

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Kafka的发送与接收数据