本文並沒有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;
 
  prote
 
程式設計客棧cted $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();
    $co程式設計客棧nf->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\ConslEdcBWumer($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(js程式設計客棧on_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);
}lEdcBW
 
?>

到此這篇關於Laravel中Kafka的使用詳解的文章就介紹到這了,更多相關Laravel中Kafka內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！