tp6中kafka使用
浏览量:87
<?php
namespace app\util;
use \RdKafka\Conf;
use \RdKafka\Producer;
class Kafka{
public $kafkaConfig;
public $kafkaResource;
//初始化
public function __construct(){
$this->connect();
}
/**
* 连接
*/
private function connect(){
$this->kafkaConfig = new Conf();
$this->kafkaConfig->set('log_level', (string) LOG_DEBUG);
$this->kafkaConfig->set('debug', 'all');
$this->kafkaResource = new Producer($this->kafkaConfig);
$this->kafkaResource->addBrokers("127.0.0.1:9092");
}
/**
* 发布消息
*/
public function publishMessage($message = ""){
$topic = $this->kafkaResource->newTopic("EASY_DATA");
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);
$this->kafkaResource->poll(0);
$this->kafkaResource->flush(10000);
return true;
}
/**
* 销毁
*/
public function __destruct(){
// $this->channel->close();
// $this->connection->close();
}
}
kafka使用代码如上
<?php
namespace app\service;
use app\util\Kafka;
class KafkaService extends \think\Service
{
/**
* 注册服务
*/
public function register()
{
$this->app->bind('kafka', Kafka::class);
}
/**
* 执行服务
*/
public function boot() {
}
}
kafka容器服务
<?php
use app\AppService;
// 系统服务定义文件
// 服务在完成全局初始化之后执行
return [
AppService::class,
'\app\service\KafkaService',
];
kafka注册服务
public function test(Request $request){
$res = app('kafka')->publishMessage('{"driver":2,"user_id":3,"ip":"127.0.0.1","type":7,"joinid":1}');
var_dump($res);
}
感谢支持与鼓励～