ITKeyword,专注技术干货聚合推荐

注册 | 登录

0.8.2kafka集群配置

u011308691 分享于 2016-07-21

推荐:kafka 生产者消费者 api接口

生产者 import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig;

2020腾讯云“6.18”活动开始了!!!(巨大优惠重现!4核8G,5M带宽 1999元/3年),
地址https://cloud.tencent.com/act/cps/redirect?redirect=1059

2020阿里云最低价产品入口,含代金券(新老用户有优惠),
地址https://www.aliyun.com/minisite/goods

hadoop,spark,kafka交流群:459898801

1,kafka配置

kafka官网的配置参数基本已经满足大家的需求了。

推荐:kafka集群扩容(Topic迁移)

文章新地址 文章开始前推荐使用 kafka-manager 工具,可以实时查看kafka同步状态,broker实时读写流量,topic分区情况等信息,提供重选最优leader等操作。 kafka

# Replication configurations
num.replica.fetchers=4
replica.fetch.max.bytes=1048576
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.socket.timeout.ms=30000
replica.socket.receive.buffer.bytes=65536
replica.lag.time.max.ms=10000
replica.lag.max.messages=4000

controller.socket.timeout.ms=30000
controller.message.queue.size=10

# Log configuration
num.partitions=8
message.max.bytes=1000000
auto.create.topics.enable=true
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.hours=168
log.flush.interval.ms=10000
log.flush.interval.messages=20000
log.flush.scheduler.interval.ms=2000
log.roll.hours=168
log.retention.check.interval.ms=300000
log.segment.bytes=1073741824

# ZK configuration
zookeeper.connection.timeout.ms=6000
zookeeper.sync.time.ms=2000

# Socket server configuration
num.io.threads=8
num.network.threads=8
socket.request.max.bytes=104857600
socket.receive.buffer.bytes=1048576
socket.send.buffer.bytes=1048576
queued.max.requests=16
fetch.purgatory.purge.interval.requests=100
producer.purgatory.purge.interval.requests=100

kafka集群配置

1,修该broker id

broker.id [阿拉伯数字,集群中不能重复]

2,zookeeper的地址列表

zookeeper.connect=ukafka-rmicyl-1-bj04.service:2181,ukafka-rmicyl-2-bj04.service:2181,ukafka-rmicyl-3-bj04.service:2181

3,广播参数

假如broker所在为内网环境并经过nat,consumer和producer所在网络在nat外部,需要配置这两个参数为nat网关ip和响应端口。kafka目前还不支持,内外网(consumer和producer一个在内网一个在外网)。

# Hostname the broker will advertise to producers and consumers. If not set, it uses the
# value for "host.name" if configured.  Otherwise, it will use the value returned from
# java.net.InetAddress.getCanonicalHostName().
advertised.host.name=<hostname routable by clients>
# The port to publish to ZooKeeper for clients to use. If this is not set,
# it will publish the same port that the broker binds to.
advertised.port=<port accessible by clients>

Kafka 0.10.0 这个版本参数如下:

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = security_protocol://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://your.host.name:9092

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://your.host.name:9092

kafka的生产者demo

props.put("metadata.broker.list", "ip1:9092,ip2:9092,ip3:9092");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("batch.size", "307200");
props.put("linger.ms", "10");
props.put("producer.type", "async");
config = new ProducerConfig(props);
producer = new Producer<String, String>(config);
String key=hell0;
String value=hi;
for(int i=0;i<10i++){
KeyedMessage<String, String> msg = new KeyedMessage<String, String>(topic, key+Integer.toString(i), value+Integer.toString(i));
producer.send(msg);
}

消费者demo


import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConsumerGroupExample {
    private final ConsumerConnector consumer;
    private final String topic;
    private  ExecutorService executor;

    public ConsumerGroupExample(String a_zookeeper, String a_groupId, String a_topic) {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                createConsumerConfig(a_zookeeper, a_groupId));
        this.topic = a_topic;
    }

    public void shutdown() {
        if (consumer != null) consumer.shutdown();
        if (executor != null) executor.shutdown();
        try {
            if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
                System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");
            }
        } catch (InterruptedException e) {
            System.out.println("Interrupted during shutdown, exiting uncleanly");
        }
   }

    public void run(int a_numThreads) {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(a_numThreads));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

        // now launch all the threads
        //
        executor = Executors.newFixedThreadPool(a_numThreads);

        // now create an object to consume the messages
        //
        int threadNumber = 0;
        for (final KafkaStream stream : streams) {
            executor.submit(new ConsumerTest(stream, threadNumber));
            threadNumber++;
        }
    }

    private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
        Properties props = new Properties();
        props.put("zookeeper.connect", a_zookeeper);
        props.put("group.id", a_groupId);
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");

        return new ConsumerConfig(props);
    }

    public static void main(String[] args) {
        String zooKeeper = args[0];
        String groupId = args[1];
        String topic = args[2];
        int threads = Integer.parseInt(args[3]);

        ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic);
        example.run(threads);

        try {
            Thread.sleep(10000);
        } catch (InterruptedException ie) {

        }
        example.shutdown();
    }
}

推荐:kafka集群操作指南

一单机版安装 二集群安装 三集群启停操作 四topic相关的操作 五某个broker挂掉本机器可重启 六某个broker挂掉且无法重启需要其它机器代替 七扩容 八数据迁移 九

hadoop,spark,kafka交流群:459898801 1,kafka配置 kafka官网的配置参数基本已经满足大家的需求了。 # Replication configurationsnum.replica.fetchers=4replica.fetch.max.bytes=1048576r

相关阅读排行


相关内容推荐

最新文章

×

×

请激活账号

为了能正常使用评论、编辑功能及以后陆续为用户提供的其他产品,请激活账号。

您的注册邮箱: 修改

重新发送激活邮件 进入我的邮箱

如果您没有收到激活邮件,请注意检查垃圾箱。