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, $value , $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;
}
}
本文由:xiaoshu168 作者:xiaoshu发表,转载请注明来源!