26.3. 入门例子

26.3.1. 订阅例子

package cn.netkiller.kafka.test;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaConsumerExample {

	private static KafkaConsumer<String, String> consumer;

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Properties props = new Properties();
		props.put("bootstrap.servers", "kafka.netkiller.cn:9092");
		props.put("group.id", "test");
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList("test"));
		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
		}
	}

}

测试方法

root@netkiller ~ % /srv/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>Helloworld

下面详细讲解上面的程序。首先我们通过Properties文件来配置消费属性。

		Properties props = new Properties();
		props.put("bootstrap.servers", "kafka.netkiller.cn:9092");
		props.put("group.id", "test");
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

然后订阅TOPIC,从那个TOPIC 读取消息,Kafka 可以同时订阅多个 TOPIC。下面的例子中,同时订阅了foo和bar两个topic:

consumer.subscribe(Arrays.asList("foo", "bar"));

取出消息(消费消息),通过循环调用poll方法,从队列中取出消息。

		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
		}		

poll方法里面的参数是等待消息的时间(Long类型),如果队列里面有消息,会立马返回,如果没有,会等待指定的时间然后返回。


26.3.2. 发布例子

package cn.netkiller.kafka.test;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka.netkiller.cn:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for(int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i)));

        producer.close();
    }
}		

原文出处:Netkiller 系列 手札
本文作者:陈景峯
转载请与作者联系,同时请务必标明文章原始出处和作者信息及本声明。

时间: 2024-12-23 09:08:08

26.3. 入门例子的相关文章

Mybatis入门例子

Mybatis是轻量级的持久化框架,的确上手非常快. Mybatis大体上的思路就是由一个总的config文件配置全局的信息,比如mysql连接信息等.然后再mapper中指定查询的sql,以及参数和返回值. 在Service中直接调用这个mapper即可. 依赖的jar包 <dependency> <groupId>org.mybatis</groupId> <artifactId>mybatis</artifactId> <versio

cocos2d-x Box2d 入门例子

1.效果图(动态的gif见底部):     还记得这篇文章吗?cocos2d-x Touch 事件应用的一个例子 球跟球之间的碰撞,球跟壁的碰撞使用了简单的向量.球最终停止下来也是因为向量不断地变小.这次我们要改用物理引擎Box2d来让球碰撞!先下载这个项目,打开Release.win32/Air Hockey.win32.exe先来个试玩吧.   2.Box2d如何在cocos2d-x中使用简单介绍   Box2d引擎中核心的是一个叫b2World的东西,它会创建一些b2Body的物体,当然这

一个经典的ADO.NET入门例子

ado 众所周知,ADO.NET相对于ADO的最大优势在于对于数据的更新修改可以在与数据源完全断开联系的情况下进行,然后再把数据更新情况传回到 数据源.这样大大减少了连接过多对于数据库服务器资源的占用.下面是我在<ADO.NET实用指南>这本书上看到的一个例子,比较清楚的讲解 了ADO.NET的使用方法.Imports System.Data.SqlClientImports System.DataImports System.Data.Common Public Class Form1 In

一个经典的ADO.NET入门例子_ASP.NET

众所周知,ADO.NET相对于ADO的最大优势在于对于数据的更新修改可以在与数据源完全断开联系的情况下进行,然后再把数据更新情况传回到  数据源.这样大大减少了连接过多对于数据库服务器资源的占用.下面是我在<ADO.NET实用指南>这本书上看到的一个例子,比较清楚的讲解  了ADO.NET的使用方法.  Imports System.Data.SqlClient  Imports System.Data  Imports System.Data.Common  Public Class For

Spring+Struts+Hibernate入门例子Wiring Step By Step

wiring下载点http://www.onjava.com/onjava/2004/04/07/examples/wiring.zip 说重点吧,主要是注意几点! 一 前期工作1. 下载Xdoclet,DB2Java和Proxool,并把Proxool lib目录下的.jar拷到wiring WEB-INF下的lib目录. 把相应的Spring,Hibernate,Struts要用到的包也拷进lib目录,如aopalliance.jar;                   cglib-ful

Delphi类的入门例子(9): 获取对象的 RTTI 属性与事件的函数

unit Unit1; interface uses Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms, Dialogs, StdCtrls, xmldom, XMLIntf, XMLBrokr, msxmldom, XMLDoc; type TForm1 = class(TForm) Button1: TButton; Button2: TButton; Memo1: TMemo; Memo2:

Delphi类的入门例子(8): 遍历窗体中所有控件的函数

//显示窗体中所有控件的函数 function GetCtrls(Control: TWinControl; List: TStringList): Boolean; var i: Integer; obj: TWinControl; begin for i := 0 to Control.ControlCount-1 do begin obj := TWinControl(Control.Controls[i]); List.Add(obj.Name); //如果控件中包含其他控件, 则递归调

Delphi类的入门例子(7): 遍历窗体的所有父类

unit Unit1; interface uses Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms, Dialogs, StdCtrls; type TForm1 = class(TForm) Button1: TButton; Button2: TButton; Button3: TButton; Button4: TButton; procedure Button1Click(Sender:

Delphi类的入门例子(4): property

//类单元unit Person; interface type TPerson = class(TObject) private FName: string; FAge: Integer; public procedure SetName(const strName: string); procedure SetAge(const intAge: Integer); property Name: string read FName write SetName; property Age: In