docker启动kafka

docker-composer.yml

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    volumes:
      - ./data:/data
    ports:
      - "2181:2181"
       
  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 192.168.1.102 
      KAFKA_MESSAGE_MAX_BYTES: 2000000
      KAFKA_CREATE_TOPICS: "Topic1:1:3,Topic2:1:1:compact"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - ./kafka-logs:/kafka
      - /var/run/docker.sock:/var/run/docker.sock
 
  kafka-manager:
    image: sheepkiller/kafka-manager
    ports:
      - 9020:9000
    environment:
      ZK_HOSTS: zookeeper:2181

docker exec -it 容器Id /bin/bash

#进入到如下目录，进行操作测试

cd /opt/kafka_2.13-2.8.1/bin

通过生产者脚本创建test_zp主题，消费者去监听消费。

#生产者
./kafka-console-producer.sh --broker-list localhost:9092 --topic test_zp


#消费者

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_zp --from-beginning

set('log_level', LOG_DEBUG);
//$conf->set('debug', 'all');
$rk = new RdKafkaProducer($conf);
$rk->addBrokers("172.30.0.10:9092");
$topic = $rk->newTopic("test_zp");
$topic->produce(RD_KAFKA_PARTITION_UA, 0, "生成的消息");

setLogLevel(LOG_DEBUG);
$rk->addBrokers("192.168.150.53:9092");
$topic = $rk->newTopic("test_zp");

$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
$n = 0;
$curOffset = 11; //业务逻辑处理完成后，更新offset,避免重复处理

while (true) {
    $n++;
    var_dump('rep:' . $n);
    sleep(2);
    if ($n > 20) break;

    $msg = $topic->consume(0, 1000);

	if (null === $msg) {
		continue;
	} elseif ($msg->err) {
		echo $msg->errstr(), "n";
		break;
	} else {
	    //
        if ($msg->offset > $curOffset) {
            var_dump('is new');
            echo $msg->payload, "n";
        } else {
            var_dump('is_rep');
            var_dump($msg);
        }
	}
}

die;

分享好文：

ubuntu上安卓php-rdkafka可通过此链接：