PHP implements open source Pulsar real-time data processing

With the development of Internet technology and the explosive growth of data volume, processing massive data has become one of the problems that today's Internet companies must face. Traditional data processing solutions, especially batch processing solutions, can no longer meet the needs of real-time and high availability. At this time, real-time data processing has become one of the best solutions. As a developer, how to handle large-scale data elegantly and efficiently is also a topic we must pay attention to.

Pulsar is a real-time data processing framework open sourced by Yahoo. It uses a layered architecture to make data processing more efficient and scalable. It supports multiple client languages, including Java, Python, Ruby, and PHP. As a very popular language, PHP has simple syntax and low learning curve. It has become one of the preferred languages ​​for many enterprises to develop real-time data processing applications. This article will introduce how to use PHP to implement real-time data processing of open source Pulsar.

Preparation

Before you start using Pulsar, you need to download and install Pulsar. Relevant software packages and documentation can be obtained from Pulsar's official website and installed on the local machine or on a node in the cluster for local development and testing.

During the PHP development process, you need to use the client SDK pulsar-client-php. It can be installed through tools such as Composer. The specific process is as follows:

// 安装pulsar-client-php
composer require apache/pulsar

After the installation is completed, the following is the basic configuration of how to use Pulsar.

use ApachePulsarAuthenticationAuthenticationFactory;
use ApachePulsarClientBuilder;
use ApachePulsarProducerConfiguration;
use ApachePulsarSerializationSerialization;

// 配置生产者的信息
$clientBuilder = new ClientBuilder();
$clientBuilder->setServiceUrl('pulsar://localhost:6650');
$clientBuilder->setAuthentication(
    AuthenticationFactory::token('your-token-string')
);

$producerConf = new ProducerConfiguration();
$producerConf->setTopic('your-topic-name');
$producerConf->setSendTimeout(3000);
$producerConf->setSerialization(Serialization::JSON);

// 创建生产者实例
$producer = $clientBuilder->build()->createProducer($producerConf);
$producer->send('your message');

In the above code, we first create the Pulsar producer through the ClientBuilder class. When creating a producer, we need to set the setServiceUrl method to specify the URL of the Pulsar Service, and the setAuthentication method to perform authentication. In addition, you need to set the producer's configuration information, such as topic, timeout, etc.

Usage of Pulsar

Pulsar provides two basic components, Producer and Consumer, to implement real-time data processing. Producer is used to send data to the specified Pulsar topic, while Consumer consumes data from the topic. Below we will introduce in detail how to use these two components to complete real-time data processing.

Producer

First, we create a Producer instance through the following steps:

// 导入命名空间
use ApachePulsarClientBuilder;

// 创建Pulsar client实例
$clientBuilder = new ClientBuilder();
$client = $clientBuilder->serviceUrl('pulsar://localhost:6650')->build();

// 创建Producer对象
$producer = $client->createProducer(
    [
        'topic' => 'your-topic',
    ]
);

When creating a producer, you need to set the Pulsar topic to which the producer belongs. In addition, there are other options, such as "producerName", "initialSequenceId", "sendTimeout", etc. These options can be configured as needed.

Let's take a look at how to send a message to the Pulsar topic:

// 对Pulsar topic发送消息
$result = $producer->send('your-message');

send method returns a MessageId object. If the message has been sent before, the corresponding MessageId is returned. If the message fails to be sent, a PulsarClientException exception is thrown.

Consumer

Like the producer, the creation of the Pulsar Consumer is also divided into multiple steps.

// 导入命名空间
use ApachePulsarClientBuilder;

// 创建Pulsar client实例
$clientBuilder = new ClientBuilder();
$client = $clientBuilder->serviceUrl('pulsar://localhost:6650')->build();

// 创建Consumer对象
$consumer = $client->subscribe(
    [
        'topic' => 'your-topic',
        'subscriptionName' => 'your-subscription-name',
    ]
);

When creating a Consumer, we need to set the Pulsar topic and subscription name of the subscription. There are other options, such as setting "receiverQueueSize", "ackTimeout", "subscriptionType", etc.

Below we will see how to get messages from the specified Pulsar topic:

// 从topic中消费消息
$message = $consumer->receive();

// 对消息进行处理
echo 'Received message with ID: ' . $message->getMessageId() . PHP_EOL;

// markAsReceived表示通知Pulsar这条消息已经被处理
$consumer->acknowledge($message);

When calling the receive() method, the program will remain in a waiting state until there is Messages are returned from the specified Pulsar topic. When a message is returned, the program will continue to execute and process the message.

After calling the acknowledge() method, Pulsar will delete the message from the queue of the subscription. If the acknowledge() method is not called, the message will remain in the queue until the message expires (default is 1 hour).

Summary

In this article, we introduced how to use PHP to implement real-time data processing of open source Pulsar. We started by setting up the Pulsar environment and described step by step how to use Pulsar's Producer and Consumer components to implement real-time data processing.

Pulsar adopts a layered architecture and can well support large-scale real-time data processing. At present, Pulsar has been used by many Internet companies, such as Alibaba, Meituan, Baidu, etc.

We believe that by studying the content introduced in this article, you can already understand how to use PHP and Pulsar to be more efficient and elegant in real-time data processing.

The above is the detailed content of PHP implements open source Pulsar real-time data processing. For more information, please follow other related articles on the PHP Chinese website!