wh0426

浏览: 56297 次
性别:
来自: 深圳

最近访客更多访客>>

brandNewUser

xiangkun

leimingchao

zx4866123

博主相关

博客

微博

相册

留言

关于我

博客专栏

: 架构师的知识与实践
浏览量：56298

文章分类

社区版块

存档分类

社区电商系统架构之消息队列篇：kafka的实验

博客分类：

架构设计

kafka 队列

1.kafka安装

tar -xzf kafka_2.10-0.8.2.1.tgz
cd kafka_2.10-0.8.2.1

2.启动

主机环境

top - 11:32:15 up 118 days, 8:18, 4 users, load average: 1.97, 3.45, 4.45
Tasks: 152 total, 1 running, 151 sleeping, 0 stopped, 0 zombie
Cpu0 : 15.0%us, 13.0%sy, 0.0%ni, 68.0%id, 3.0%wa, 0.0%hi, 1.0%si, 0.0%st
Cpu1 : 26.2%us, 15.0%sy, 0.0%ni, 56.8%id, 0.0%wa, 0.0%hi, 2.0%si, 0.0%st
Cpu2 : 62.7%us, 13.7%sy, 0.0%ni, 18.3%id, 0.0%wa, 0.0%hi, 5.3%si, 0.0%st
Cpu3 : 16.6%us, 14.0%sy, 0.0%ni, 66.1%id, 2.7%wa, 0.0%hi, 0.7%si, 0.0%st

启动kafka zookeeper进程

bin/zookeeper-server-start.sh config/zookeeper.properties &

启动leader
bin/kafka-server-start.sh config/server.properties &

启动follower 1

bin/kafka-server-start.sh config/server-1.properties &

启动follower 2

bin/kafka-server-start.sh config/server-2.properties &

3.实验准备

1.创建kafka主题队列
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic page_visits1

2.查看Kafka主题队列是否加入数据
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic page_visits1 --from-beginning

4.常见异常

解决FailedToSendMessageException问题。
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
修改config/server.properties中 host.name=主机地址

5.写入主题消息

TestProducer

package kafka.client;


import java.util.*;


import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class TestProducer {
  public static void main(String[] args) {
    long events = 1000000;
    Random rnd = new Random();

    Properties props = new Properties();
    //props.put("zk.connect", "192.168.161.73:2181");
    props.put("metadata.broker.list", "192.168.161.73:9092");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
   // props.put("zookeeper.session.timeout.ms", "400000");
    props.put("partitioner.class", "kafka.client.SimplePartitioner");
    //props.put("request.required.acks", "1");

    ProducerConfig config = new ProducerConfig(props);

    Producer<String, String> producer = new Producer<String, String>(config);

    long begin= System.currentTimeMillis();
    for (long nEvents = 0; nEvents < events; nEvents++) {
       long runtime = new Date().getTime(); 
       String ip = "192.168.2." + rnd.nextInt(255);
       String msg = runtime + ",www.example.com," + ip;
       KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits1", runtime+"",ip, msg);
       producer.send(data);
    }
    long end= System.currentTimeMillis();
    System.out.println("cost time:"+(end-begin));
    producer.close();
  }
}

6.消费主题消息

SimplePartitioner 可选

package kafka.client;
import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;
 
public class SimplePartitioner implements Partitioner {
    public SimplePartitioner (VerifiableProperties props) {
 
    }
 
    public int partition(Object key, int a_numPartitions) {
        int partition = 0;
        String stringKey = (String) key;
        int offset = stringKey.lastIndexOf('.');
        if (offset > 0) {
           partition = Integer.parseInt( stringKey.substring(offset+1)) % a_numPartitions;
        }
       return partition;
  }
 
}

ConsumerGroupExample

package kafka.client;
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;
import java.util.concurrent.TimeUnit;
 
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 = "192.168.161.73:2181";
        String groupId = "testgroup";
        String topic = "page_visits1";
        int threads = Integer.parseInt("10");
 
        ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic);
        example.run(threads);
 
        try {
            Thread.sleep(10000);
        } catch (InterruptedException ie) {
 
        }
        example.shutdown();
    }
}

ConsumerTest

package kafka.client;


import java.util.concurrent.Callable;


import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

public class ConsumerTest implements Runnable {
  private KafkaStream m_stream;
  private int m_threadNumber;

  public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
    m_threadNumber = a_threadNumber;
    m_stream = a_stream;
  }

  public void run() {
      long begin= System.currentTimeMillis();
    ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
    System.out.println( "stream.iterator cost time="+(begin-begin));
   // int s= m_stream.size();
    long begin1= System.currentTimeMillis();
    int s=0;
    String val="";
    while (it.hasNext()){
     val=String.valueOf(it.next().message());
    if(s%10000==0) {
       System.out.println("Thread " + m_threadNumber + ": "+val );}
      s++;
    }
   
    System.out.println("Shutting down Thread: " + m_threadNumber);
    long end= System.currentTimeMillis();
    System.out.println("thread "+m_threadNumber+ " cost time="+(end-begin)+",queue counter="+s);
  }
}

7.测试

1.运行TestProducer，向page_visits1主题发布消息。

第一次运行

./run-producter.sh
[INFO] starting ...
SERVER_HOME=/root/kafka-test
-Xms2048M -Xmx2048M -XX:MaxPermSize=1024M -XX:PermSize=512M -XX:+UseG1GC
BOOTCMD=java -cp /root/kafka-test/conf:/root/kafka-test/lib:/root/kafka-test/lib/* kafka.client.TestProducer
java -cp /root/kafka-test/conf:/root/kafka-test/lib:/root/kafka-test/lib/* kafka.client.TestProducer
log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
cost time:150936 。即150秒.

第二次运行：

2.运行ConsumerGroupExample，消息page_visits1主题消息

第一次运行：

./run-consumer.sh
[INFO] starting ...
SERVER_HOME=/root/kafka-test
-Xms2048M -Xmx2048M -XX:MaxPermSize=1024M -XX:PermSize=512M -XX:+UseG1GC
BOOTCMD=java -cp /root/kafka-test/conf:/root/kafka-test/lib:/root/kafka-test/lib/* kafka.client.ConsumerGroupExample
java -cp /root/kafka-test/conf:/root/kafka-test/lib:/root/kafka-test/lib/* kafka.client.ConsumerGroupExample
log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Shutting down Thread: 9
thread 9 cost time=10029,queue counter=98576
Shutting down Thread: 4
thread 4 cost time=10045,queue counter=97970
Shutting down Thread: 0
thread 0 cost time=10048,queue counter=101707
Shutting down Thread: 8
thread 8 cost time=10030,queue counter=101870
Shutting down Thread: 1
thread 1 cost time=10048,queue counter=97799
Shutting down Thread: 7
thread 7 cost time=10038,queue counter=97865
Shutting down Thread: 5
Shutting down Thread: 6
thread 5 cost time=10038,queue counter=102579
Shutting down Thread: 2
thread 2 cost time=10047,queue counter=102181
thread 6 cost time=10045,queue counter=97463
Shutting down Thread: 3
thread 3 cost time=10047,queue counter=101990

top - 11:36:04 up 118 days, 8:22, 4 users, load average: 0.80, 2.21, 3.76
Tasks: 151 total, 1 running, 150 sleeping, 0 stopped, 0 zombie
Cpu0 : 13.3%us, 46.7%sy, 0.0%ni, 40.0%id, 0.0%wa, 0.0%hi, 0.0%si, 0.0%st
Cpu1 : 16.2%us, 36.8%sy, 0.0%ni, 47.1%id, 0.0%wa, 0.0%hi, 0.0%si, 0.0%st
Cpu2 : 17.3%us, 37.3%sy, 0.0%ni, 44.0%id, 0.0%wa, 0.0%hi, 1.3%si, 0.0%st
Cpu3 : 15.3%us, 37.5%sy, 0.0%ni, 44.4%id, 0.0%wa, 0.0%hi, 2.8%si, 0.0%st

完成所有消息消费，约100秒钟

第二次运行：

./run-consumer.sh
[INFO] starting ...
SERVER_HOME=/root/kafka-test
-Xms2048M -Xmx2048M -XX:MaxPermSize=1024M -XX:PermSize=512M -XX:+UseG1GC
BOOTCMD=java -cp /root/kafka-test/conf:/root/kafka-test/lib:/root/kafka-test/lib/* kafka.client.ConsumerGroupExample
java -cp /root/kafka-test/conf:/root/kafka-test/lib:/root/kafka-test/lib/* kafka.client.ConsumerGroupExample
log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Shutting down Thread: 9
Shutting down Thread: 1
thread 9 cost time=10026,queue counter=0
Shutting down Thread: 4
Shutting down Thread: 0
Shutting down Thread: 8
thread 0 cost time=10037,queue counter=0
thread 4 cost time=10028,queue counter=0
Shutting down Thread: 7
thread 7 cost time=10027,queue counter=0
thread 1 cost time=10027,queue counter=0
thread 8 cost time=10027,queue counter=0
Shutting down Thread: 2
thread 2 cost time=10034,queue counter=0
Shutting down Thread: 6
thread 6 cost time=10028,queue counter=0
Shutting down Thread: 5
thread 5 cost time=10028,queue counter=0
Shutting down Thread: 3
thread 3 cost time=10034,queue counter=0

无任何消息消费。居然发现cost time与第一次相同。说明每个线程耗用10s，是线程run从开始到线程结束，所需要的时间。

实验代码地址：

https://code.csdn.net/wh0426/kafkademo/tree/master

8.结论

1.Kafka队列消息插入速度非常快，发布100W请求，耗时150秒。

2.Kafka队列消息消费也非常快，消费100W请求，耗时100秒

3.kfaka队列支持Partition分区功能，默认按key进行hash分区，自定义扩展实现Partitioner接口，并指定partitioner.class属性配置为自定义扩展类。

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic page_visits1

Topic:page_visits1PartitionCount:10ReplicationFactor:1Configs:
Topic: page_visits1Partition: 0Leader: 2Replicas: 2Isr: 2
Topic: page_visits1Partition: 1Leader: 0Replicas: 0Isr: 0
Topic: page_visits1Partition: 2Leader: 2Replicas: 2Isr: 2
Topic: page_visits1Partition: 3Leader: 0Replicas: 0Isr: 0
Topic: page_visits1Partition: 4Leader: 2Replicas: 2Isr: 2
Topic: page_visits1Partition: 5Leader: 0Replicas: 0Isr: 0
Topic: page_visits1Partition: 6Leader: 2Replicas: 2Isr: 2
Topic: page_visits1Partition: 7Leader: 0Replicas: 0Isr: 0
Topic: page_visits1Partition: 8Leader: 2Replicas: 2Isr: 2
Topic: page_visits1Partition: 9Leader: 0Replicas: 0Isr: 0

4.后续需要测试单点故障，当一台leader kafka 失败后，剩余的follower是否可以继续接管任务。

分享到：

tomcat7连接器配置，bio与nio连接器 | 社区电商系统架构之服务治理篇：dubbo的实 ...

2015-04-09 09:35
浏览 3943
评论(0)
分类:企业架构
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

博客专栏

文章分类

社区版块

存档分类

最新评论

社区电商系统架构之消息队列篇：kafka的实验

1.kafka安装

2.启动

3.实验准备

4.常见异常

5.写入主题消息

6.消费主题消息

7.测试

8.结论

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

博客专栏

文章分类

社区版块

存档分类

最新评论

社区电商系统架构之消息队列篇：kafka的实验

1.kafka安装

2.启动

3.实验准备

4.常见异常

5.写入主题消息

6.消费主题消息

7.测试

8.结论

评论

发表评论

相关推荐

tomcat msm部署

社区电商系统架构之会话篇：sso单点登录会话介绍

电商系统架构总论篇

社区电商系统架构之技术架构验证篇2

社区电商系统架构之jvm远程监控

社区电商系统架构之服务治理篇：dubbo的实验

社区电商之加密码模块JceSecurity内存不释放问题

disruptor使用示例

redis分布式内存锁：余量扣除示例

netty实现http api功能

最近访客更多访客>>