1.安装组件
composer require nmred/kafka-php
2.创建者kafka类
<?php
namespace App\Http\Services;
use Kafka;
class KafkaService
{
public function __construct()
{
date_default_timezone_set('PRC');
}
/*
* Produce
*/
public function Producer($topic,$value,$url)
{
$config = \Kafka\ProducerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList($url);
$config->setBrokerVersion('1.0.0');
$config->setRequiredAck(1);
$config->setIsAsyn(false);
$config->setProduceInterval(500);
$producer = new \Kafka\Producer(function () use($value,$topic){
return [
[
'topic' => $topic,'value' => $value,'key' => '',],];
});
$producer->success(function ($result){
return "success";
});
$producer->error(function ($errorCode){
var_dump($errorCode);
});
$producer->send(true);
}
/*
* Consumer
*/
public function consumer($group,$topics,$url){
$config = \Kafka\ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(500);
$config->setMetadataBrokerList($url);
$config->setGroupId($group);
$config->setBrokerVersion('1.0.0');
$config->setTopics([$topics]);
$config->setOffsetReset('earliest');
$consumer = new \Kafka\Consumer();
$consumer->start(function($topic,$part,$message) {
echo "receive a message...\n";
app('consumerKafka')->consumerData($message['message']['value']); //你的接收处理逻辑
var_dump($message['message']['value']);
});
}
}
3.生产者
<?php
namespace App\Http\Services;
use App\Http\Services\KafkaService;
class ProduceService
{
public function produce()
{
$topic = env('topic_test'); //配置在env中
$url = env('kafka_url_test'); //配置在env中
$value =
[
'code' => 'test','data_type' => 'personal','action' => 'update','data' =>
[
'id' => 1,'name' => 'tom','gender' => 2
],'redirect_url' => '','operator' => 'system',];
$value = json_encode ($value,JSON_FORCE_OBJECT );
$kafka = new KafkaService();
$kafka->Producer($topic,$url);
}
}
4.执行 php artisan consumer:kafka 消费消息
<?php
namespace App\Console\Commands;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Redis;
class ConsumerKafka extends Command
{
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'consumer:kafka';
/**
* The console command description.
*
* @var string
*/
protected $description = '处理异步kafka消息';
/**
* Create a new command instance.
*/
public function __construct()
{
parent::__construct();
}
/**
* Execute the console command.
*
* @return mixed
*/
public function handle()
{
//开始监听消息.
app('kafkaService')->consumer($group=env('KAFKA_GROUP'),$topics =env('KAFKA_TOPIC'),$url=env('KAFKA_URL'));
return $this;
}
}