KafkaTemplate的使用

	<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="${kafka.zookeeper.connect}" zk-connection-timeout="6000" zk-session-timeout="400" zk-sync-time="200" />	
	
	<bean id="zookeeperConfiguration" class="org.springframework.integration.kafka.core.ZookeeperConfiguration">
		<constructor-arg name="zookeeperConnect" ref="zookeeperConnect" />
	</bean>
	
	<bean id="defaultConnectionFactory" class="org.springframework.integration.kafka.core.DefaultConnectionFactory">
		<constructor-arg name="configuration" ref="zookeeperConfiguration" />
	</bean>

	<bean id="kafkaTemplate" class="org.springframework.integration.kafka.core.KafkaTemplate">
		<constructor-arg name="connectionFactory" ref="defaultConnectionFactory" />
	</bean>

 

package com.vti.test;

import java.util.Collection;
import java.util.Collections;
import java.util.List;

import javax.annotation.Resource;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.integration.kafka.core.FetchRequest;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.KafkaMessageBatch;
import org.springframework.integration.kafka.core.KafkaTemplate;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.core.Result;
import org.springframework.integration.kafka.serializer.common.StringDecoder;
import org.springframework.integration.kafka.util.MessageUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath:beans.xml" })
public class KafkaTemplateTest {

	@Resource
	private KafkaTemplate kafkaTemplate;

	@Test
	public void receive() {

		String topic = "test";
		long offset = 0;
		int max = 1000;

		Collection<Partition> partitions = kafkaTemplate.getConnectionFactory().getPartitions(topic);

		while (partitions.iterator().hasNext()) {

			Partition partition = partitions.iterator().next();

			Result<KafkaMessageBatch> receive = kafkaTemplate
					.receive(Collections.singleton(new FetchRequest(partitions.iterator().next(), offset, max)));

			List<KafkaMessage> messages = receive.getResult(partition).getMessages();

			if (messages.size() > 0) {

				for (KafkaMessage kafkaMessage : messages) {

					String msg = MessageUtils.decodePayload(kafkaMessage, new StringDecoder());

					offset = kafkaMessage.getMetadata().getNextOffset();

					System.out.println(msg);
				}

			} else {
				try {
					Thread.sleep(3*1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				continue;
			}
		}
	}
}