主页 > 知识库 > Laravel中Kafka的使用详解

Laravel中Kafka的使用详解

热门标签:常州地图标注服务商 衡水外呼系统平台 新河科技智能外呼系统怎么样 福州人工外呼系统哪家强 百度商铺地图标注 地图标注平台怎么给钱注册 注册400电话申请 安装电销外呼系统 钉钉打卡地图标注

本文并没有kafka的安装教程,本文是针对已经安装kafka及其配置好kafka的php拓展并且使用laravel框架进行开发项目,配置一个可供laravel框架使用的生产及消费者类.

以下代码修改自本站的YII框架关于kafka类的代码,经过测试使用在本人的项目中,可正常运行,larvael版本:5.6 代码放置larvael框架位置:app/Tools/Kafka.php

?php
namespace App\Tools;
 
use Illuminate\Config\Repository;
 
use Illuminate\Support\Facades\DB;
use Monolog\Logger;
use Monolog\Handler\StreamHandler;
 
use Illuminate\Http\Request;
 
class Kafka
{
  public $broker_list = '127.0.0.1';//配置kafka,可以用逗号隔开多个kafka
  public $topic = 'test';//管道名称
  public $partition = 0;
 
  protected $producer = null;
  protected $consumer = null;
 
  public function __construct()
  {
    if (empty($this->broker_list)) {
      throw new InvalidConfigException("broker not config");
    }
    $rk = new \RdKafka\Producer();
    if (empty($rk)) {
      throw new InvalidConfigException("producer error");
    }
    $rk->setLogLevel(LOG_DEBUG);
    if (!$rk->addBrokers($this->broker_list)) {
      throw new InvalidConfigException("producer error");
    }
    $this->producer = $rk;
  }
 
  /**
   * 生产者
   * @param array $messages
   * @return mixed
   */
  public function send($messages = [],$topic)
  {
    $topic = $this->producer->newTopic($topic);
    return $topic->produce(RD_KAFKA_PARTITION_UA, $this->partition, json_encode($messages));
  }
 
  /**
   * 消费者
   */
  public function consumer($object, $callback){
    $conf = new \RdKafka\Conf();
    $conf->set('group.id', 0);
    $conf->set('metadata.broker.list', $this->broker_list);
 
    $topicConf = new \RdKafka\TopicConf();
    $topicConf->set('auto.offset.reset', 'smallest');
 
    $conf->setDefaultTopicConf($topicConf);
 
    $consumer = new \RdKafka\KafkaConsumer($conf);
 
    $consumer->subscribe([$this->topic]);
 
    echo "waiting for messages.....\n";
    while(true) {
      $message = $consumer->consume(120*1000);
      switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
          echo "message payload....";
          $object->$callback($message->payload);
          break;
      }
      sleep(1);
    }
  }
}
?>

在控制器中如何使用:

首先再头部导入这个类:use App\Tools\Kafka;

下面是使用生产者实例:

public function test(){
 
   $topic = 'tool';//输入使用管道名称
   $data['shop_id'] = 58;
   $data['bar_code']=586;
   $data['goods_num'] = 1;
   $data['goods_unit'] = '个';
 
$Kafka = new Kafka();
$Error_Msg = $Kafka->send($data,$topic);//传入数组会自动转换json
var_dump($Error_Msg);
 
 
  }

下面是消费者实例,消费者我这里使用了的是php脚本进行的操作:

?php
 
$conf = new RdKafka\Conf();
 
$conf->set('group.id', 'myConsumerGroup');
 
$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("localhost:9092");
 
$topicConf = new RdKafka\TopicConf();
$topicConf->set('auto.commit.interval.ms', 100);
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', sys_get_temp_dir());
$topicConf->set('auto.offset.reset', 'smallest');
 
$topic = $rk->newTopic("tool", $topicConf);//读取的管道
 
// Start consuming partition 0
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
 
while (true) {
  $message = $topic->consume(0, 120*10000);
  switch ($message->err) {
    case RD_KAFKA_RESP_ERR_NO_ERROR:
    //没有错误打印信息
      $message = json_decode(json_encode($message),true);
      $data = json_decode($message['payload'],true);
      var_dump($data);
      break;
    case RD_KAFKA_RESP_ERR__PARTITION_EOF:
      echo "等待接收信息\n";
      break;
    case RD_KAFKA_RESP_ERR__TIMED_OUT:
      echo "超时\n";
      break;
    default:
      throw new \Exception($message->errstr(), $message->err);
      break;
  }
 sleep(1);
}
 
?>

到此这篇关于Laravel中Kafka的使用详解的文章就介绍到这了,更多相关Laravel中Kafka内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
  • 浅谈Laravel中使用Slack进行异常通知
  • 如何用Laravel包含你自己的帮助函数
  • 详解Laravel框架的依赖注入功能
  • php+laravel 扫码二维码签到功能
  • laravel的数据表填充器使用详解
  • laravel ajax curd 搜索登录判断功能的实现
  • laravel使用redis队列实例讲解
  • Laravel的加密解密与哈希实例讲解
  • Laravel中10个有用的用法小结
  • 分析五个Laravel Dusk的使用技巧

标签:克拉玛依 唐山 辽阳 鹤岗 鹰潭 柳州 白城 六安

巨人网络通讯声明:本文标题《Laravel中Kafka的使用详解》,本文关键词  Laravel,中,Kafka,的,使用,详解,;如发现本文内容存在版权问题,烦请提供相关信息告之我们,我们将及时沟通与处理。本站内容系统采集于网络,涉及言论、版权与本站无关。
  • 相关文章
  • 下面列出与本文章《Laravel中Kafka的使用详解》相关的同类信息!
  • 本页收集关于Laravel中Kafka的使用详解的相关信息资讯供网民参考!
  • 企业400电话

    智能AI客服机器人
    15000

    在线订购

    合计11份范本:公司章程+合伙协议+出资协议+合作协议+股权转让协议+增资扩股协议+股权激励+股东会决议+董事会决议

    推荐文章