kafka_2.11-0.8.2.1+java 生产消费程序demo示例

时间:2024-03-24 10:03:26

kafka是吞吐量巨大的一个消息系统,它是用scala写的,和普通的消息的生产消费还有所不同,写了个demo程序供大家参考。kafka的安装请参考官方文档。

首先我们需要新建一个maven项目,然后在pom中引用kafka jar包,引用依赖如下:


  1. <dependency>
  2. <groupId>org.apache.kafka</groupId>
  3. <artifactId>kafka_2.11</artifactId>
  4. <version>0.8.2.1</version>
  5. </dependency>

我们用的版本是0.8, 下面我们看下生产消息的代码:


  1. package com.telewave.kafka.util;
  2. import java.util.Properties;
  3. import kafka.javaapi.producer.Producer;
  4. import kafka.producer.KeyedMessage;
  5. import kafka.producer.ProducerConfig;
  6. /**
  7. *
  8. * Hello world!
  9. *
  10. *
  11. */
  12. public class KafkaProducer
  13. {
  14. private final Producer<String, String> producer;
  15. public final static String TOPIC = "TestTopic";
  16. private KafkaProducer() {
  17. Properties props = new Properties();
  18. // 此处配置的是kafka的端口
  19. props.put("metadata.broker.list", "192.168.168.200:9092");
  20. // 配置value的序列化类
  21. props.put("serializer.class", "kafka.serializer.StringEncoder");
  22. // 配置key的序列化类
  23. props.put("key.serializer.class", "kafka.serializer.StringEncoder");
  24. // request.required.acks
  25. // 0, which means that the producer never waits for an acknowledgement
  26. // from the broker (the same behavior as 0.7). This option provides the
  27. // lowest latency but the weakest durability guarantees (some data will
  28. // be lost when a server fails).
  29. // 1, which means that the producer gets an acknowledgement after the
  30. // leader replica has received the data. This option provides better
  31. // durability as the client waits until the server acknowledges the
  32. // request as successful (only messages that were written to the
  33. // now-dead leader but not yet replicated will be lost).
  34. // -1, which means that the producer gets an acknowledgement after all
  35. // in-sync replicas have received the data. This option provides the
  36. // best durability, we guarantee that no messages will be lost as long
  37. // as at least one in sync replica remains.
  38. props.put("request.required.acks", "-1");
  39. producer = new Producer<String, String>(new ProducerConfig(props));
  40. }
  41. void produce() {
  42. int messageNo = 1000;
  43. final int COUNT = 10000;
  44. while (messageNo < COUNT) {
  45. String key = String.valueOf(messageNo);
  46. String data = "hello kafka message " + key;
  47. producer.send(new KeyedMessage<String, String>(TOPIC, key, data));
  48. System.out.println(data);
  49. messageNo++;
  50. }
  51. }
  52. public static void main(String[] args)
  53. {
  54. new KafkaProducer().produce();
  55. }
  56. }

下面是消费端的代码实现:


  1. package com.telewave.kafka.util;
  2. import java.util.HashMap;
  3. import java.util.List;
  4. import java.util.Map;
  5. import java.util.Properties;
  6. import org.apache.kafka.clients.producer.KafkaProducer;
  7. import kafka.consumer.ConsumerConfig;
  8. import kafka.consumer.ConsumerIterator;
  9. import kafka.consumer.KafkaStream;
  10. import kafka.javaapi.consumer.ConsumerConnector;
  11. import kafka.serializer.StringDecoder;
  12. import kafka.utils.VerifiableProperties;
  13. public class KafkaConsumer {
  14. private final ConsumerConnector consumer;
  15. public KafkaConsumer() {
  16. Properties props = new Properties();
  17. // zookeeper 配置
  18. props.put("zookeeper.connect", "192.168.168.200:2181");
  19. // group 代表一个消费组
  20. props.put("group.id", "jd-group");
  21. // zk连接超时
  22. props.put("zookeeper.session.timeout.ms", "4000");
  23. props.put("zookeeper.sync.time.ms", "200");
  24. props.put("auto.commit.interval.ms", "1000");
  25. props.put("auto.offset.reset", "largest");
  26. // 序列化类
  27. props.put("serializer.class", "kafka.serializer.StringEncoder");
  28. ConsumerConfig config = new ConsumerConfig(props);
  29. consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
  30. }
  31. public void consume() {
  32. Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
  33. topicCountMap.put("TestTopic", new Integer(1));
  34. StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
  35. StringDecoder valueDecoder = new StringDecoder(
  36. new VerifiableProperties());
  37. Map<String, List<KafkaStream<String, String>>> consumerMap =
  38. consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
  39. KafkaStream<String, String> stream = consumerMap.get(
  40. "TestTopic").get(0);
  41. ConsumerIterator<String, String> it = stream.iterator();
  42. while (it.hasNext())
  43. System.out.println(it.next().message());
  44. }
  45. public static void main(String[] args) {
  46. new KafkaConsumer().consume();
  47. }
  48. }