PHP는 오픈 소스 Pulsar 실시간 데이터 처리를 구현합니다.

인터넷 기술의 발전과 데이터 양의 폭발적인 증가로 인해 대용량 데이터의 처리는 오늘날 인터넷 기업이 직면해야 할 문제 중 하나가 되었습니다. 기존 데이터 처리 솔루션, 특히 일괄 처리 솔루션은 더 이상 실시간 및 고가용성 요구 사항을 충족할 수 없습니다. 현재 실시간 데이터 처리는 최고의 솔루션 중 하나가 되었습니다. 개발자로서 대규모 데이터를 어떻게 우아하고 효율적으로 처리할 것인가 역시 우리가 주목해야 할 주제입니다.

Pulsar는 Yahoo에서 오픈 소스로 제공하는 실시간 데이터 처리 프레임워크로, 계층화된 아키텍처를 사용하여 데이터 처리를 더욱 효율적이고 확장 가능하게 만듭니다. Java, Python, Ruby 및 PHP를 포함한 여러 클라이언트 언어를 지원합니다. 매우 널리 사용되는 언어인 PHP는 구문이 간단하고 학습 곡선이 낮습니다. 이는 많은 기업에서 실시간 데이터 처리 애플리케이션을 개발하는 데 선호되는 언어 중 하나가 되었습니다. 이 기사에서는 PHP를 사용하여 오픈 소스 Pulsar의 실시간 데이터 처리를 구현하는 방법을 소개합니다.

준비

Pulsar를 사용하기 전에 먼저 Pulsar를 다운로드하고 설치해야 합니다. 관련 소프트웨어 패키지 및 문서는 Pulsar의 공식 웹사이트에서 얻을 수 있으며 로컬 개발 및 테스트를 위해 로컬 시스템이나 클러스터의 노드에 설치할 수 있습니다.

PHP 개발 프로세스 중에는 클라이언트 SDK pulsar-client-php를 사용해야 합니다. Composer 등의 도구를 통해 설치할 수 있습니다. 구체적인 과정은 다음과 같습니다.

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

설치가 완료된 후 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');

위 코드에서는 먼저 ClientBuilder 클래스를 통해 Pulsar 생산자를 생성합니다. 생산자를 생성할 때 setServiceUrl 메서드를 설정하여 Pulsar 서비스의 URL을 지정하고 setAuthentication 메서드를 설정하여 인증을 수행해야 합니다. 또한 주제, 시간 초과 등 생산자의 구성 정보를 설정해야 합니다. ClientBuilder类来创建Pulsar的生产者。在创建生产者的时候，我们需要设置setServiceUrl方法来指定Pulsar Service的URL，setAuthentication方法来进行身份验证。另外需要设置生产者的配置信息，如话题、超时等。

Pulsar的使用

Pulsar提供了Producer和Consumer两种基本的组件实现实时数据处理。Producer用于将数据发送到指定的Pulsar topic，而Consumer则从topic中消费数据。下面我们将详细介绍如何使用这两种组件来完成实时数据处理。

Producer

首先，我们通过以下步骤来创建一个Producer实例：

// 导入命名空间
use ApachePulsarClientBuilder;

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

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

在创建生产者时，需要设置生产者所属的Pulsar topic。此外，还有其他可选项，如“producerName”、“initialSequenceId”、“sendTimeout”等。这些选项可以根据需要进行配置。

下面我们来看一下如何向Pulsar topic发送消息：

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

send方法返回一个MessageId对象。如果消息之前已经发送过，则返回对应的MessageId。如果消息发送失败，则抛出PulsarClientException异常。

Consumer

与生产者一样，Pulsar Consumer的创建也是分为多个步骤。

// 导入命名空间
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',
    ]
);

在创建Consumer时，我们需要设置订阅的Pulsar topic和订阅名称。另外有其他可选项，如设置“receiverQueueSize”、“ackTimeout”、“subscriptionType”等。

下面我们将看到如何从指定的Pulsar topic中获取消息：

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

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

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

在调用receive()方法时，程序会保持等待状态，直到有消息从指定的Pulsar topic中返回。当有消息返回时，程序会继续执行，对消息进行处理。

调用acknowledge()方法后，Pulsar才会将消息从该订阅的队列中删除。如果没有调用acknowledge()

Producer

rrreee

Consumer

위 내용은 PHP는 오픈 소스 Pulsar 실시간 데이터 처리를 구현합니다.의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!