Pesan terurut adalah jenis pesan yang disediakan oleh ApsaraMQ for RocketMQ. Pesan ini diterbitkan dan dikonsumsi dalam urutan first-in-first-out (FIFO) yang ketat. Topik ini memberikan contoh kode untuk mengirim dan menerima pesan terurut menggunakan HTTP client SDK untuk PHP.
Informasi latar belakang
Pesan terurut dibagi menjadi beberapa jenis berikut:
Pesan terurut global: Jika pesan dalam topik termasuk jenis ini, pesan tersebut diterbitkan dan dikonsumsi dalam urutan FIFO.
Pesan terurut berpartisi: Jika pesan dalam topik termasuk jenis ini, pesan didistribusikan ke partisi yang berbeda menggunakan kunci sharding. Pesan dalam setiap partisi dikonsumsi dalam urutan FIFO. Kunci sharding adalah bidang kunci yang digunakan untuk mengidentifikasi partisi pada pesan terurut. Kunci sharding berbeda dari kunci pesan.
Untuk informasi lebih lanjut, lihat Pesan Terurut.
Prasyarat
Sebelum memulai, pastikan langkah-langkah berikut telah dilakukan:
Instal SDK untuk PHP. Untuk informasi lebih lanjut, lihat Persiapkan Lingkungan.
Buat sumber daya yang ingin Anda tentukan dalam kode di konsol ApsaraMQ for RocketMQ. Sumber daya tersebut mencakup instance, topik, dan grup konsumen. Untuk informasi lebih lanjut, lihat Buat Sumber Daya.
Peroleh pasangan AccessKey dari akun Alibaba Cloud Anda. Untuk informasi lebih lanjut, lihat Buat Pasangan AccessKey.
Kirim pesan terurut
Penting
Broker ApsaraMQ for RocketMQ menentukan urutan pesan berdasarkan urutan pengiriman menggunakan satu produser atau thread untuk mengirim pesan. Jika pengirim menggunakan beberapa produser atau thread secara bersamaan, urutan pesan ditentukan oleh urutan penerimaan pesan oleh broker ApsaraMQ for RocketMQ. Urutan ini mungkin berbeda dari urutan pengiriman di sisi bisnis.
Contoh kode berikut menunjukkan cara mengirim pesan terurut menggunakan HTTP client SDK untuk PHP:
<?php
require "vendor/autoload.php";
use MQ\Model\TopicMessage;
use MQ\MQClient;
class ProducerTest
{
private $client;
private $producer;
public function __construct()
{
$this->client = new MQClient(
// Titik akhir HTTP. Anda dapat memperoleh titik akhir di bagian HTTP Endpoint halaman Detail Instance di konsol ApsaraMQ for RocketMQ.
"${HTTP_ENDPOINT}",
// Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET dikonfigurasi.
// ID AccessKey yang digunakan untuk autentikasi.
getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
// Rahasia AccessKey yang digunakan untuk autentikasi.
getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
);
// Topik tempat pesan diproduksi. Anda harus membuat topik di konsol ApsaraMQ for RocketMQ.
$topic = "${TOPIC}";
// ID instance tempat topik dimiliki. Anda harus membuat instance di konsol ApsaraMQ for RocketMQ.
// Jika instance memiliki namespace, tentukan ID instance. Jika instance tidak memiliki namespace, atur parameter instanceID ke null atau string kosong. Anda dapat memperoleh namespace instance di halaman Detail Instance di konsol ApsaraMQ for RocketMQ.
$instanceId = "${INSTANCE_ID}";
$this->producer = $this->client->getProducer($instanceId, $topic);
}
public function run()
{
try
{
for ($i=1; $i<=4; $i++)
{
$publishMessage = new TopicMessage(
"hello mq! "// Isi pesan.
);
// Atribut kustom pesan.
$publishMessage->putProperty("a", $i);
// Kunci sharding yang digunakan untuk mendistribusikan pesan terurut ke partisi tertentu. Kunci sharding dapat digunakan untuk mengidentifikasi partisi. Kunci sharding berbeda dari kunci pesan.
$publishMessage->setShardingKey($i % 2);
$result = $this->producer->publishMessage($publishMessage);
print "Pengiriman pesan mq berhasil. msgId adalah:" . $result->getMessageId() . ", bodyMD5 adalah:" . $result->getMessageBodyMD5() . "\n";
}
} catch (\Exception $e) {
print_r($e->getMessage() . "\n");
}
}
}
$instance = new ProducerTest();
$instance->run();
?>
Berlangganan pesan terurut
Contoh kode berikut menunjukkan cara berlangganan pesan terurut menggunakan HTTP client SDK untuk PHP:
<?php
require "vendor/autoload.php";
use MQ\MQClient;
class ConsumerTest
{
private $client;
private $consumer;
public function __construct()
{
$this->client = new MQClient(
// Titik akhir HTTP. Anda dapat memperoleh titik akhir di bagian HTTP Endpoint halaman Detail Instance di konsol ApsaraMQ for RocketMQ.
"${HTTP_ENDPOINT}",
// Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET dikonfigurasi.
// ID AccessKey yang digunakan untuk autentikasi.
getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
// Rahasia AccessKey yang digunakan untuk autentikasi.
getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
);
// Topik tempat pesan diproduksi. Anda harus membuat topik di konsol ApsaraMQ for RocketMQ.
$topic = "${TOPIC}";
// ID grup konsumen yang Anda buat di konsol ApsaraMQ for RocketMQ.
$groupId = "${GROUP_ID}";
// ID instance tempat topik dimiliki. Anda harus membuat instance di konsol ApsaraMQ for RocketMQ.
// Jika instance memiliki namespace, tentukan ID instance. Jika instance tidak memiliki namespace, atur parameter instanceID ke null atau string kosong. Anda dapat memperoleh namespace instance di halaman Detail Instance di konsol ApsaraMQ for RocketMQ.
$instanceId = "${INSTANCE_ID}";
$this->consumer = $this->client->getConsumer($instanceId, $topic, $groupId);
}
public function ackMessages($receiptHandles)
{
try {
$this->consumer->ackMessage($receiptHandles);
} catch (\Exception $e) {
if ($e instanceof MQ\Exception\AckMessageException) {
// Jika handle pesan kedaluwarsa, broker tidak dapat menerima pengakuan (ACK) untuk pesan dari konsumen.
printf("Kesalahan Pengakuan, RequestId:%s\n", $e->getRequestId());
foreach ($e->getAckMessageErrorItems() as $errorItem) {
printf("\tReceiptHandle:%s, KodeKesalahan:%s, PesanKesalahan:%s\n", $errorItem->getReceiptHandle(), $errorItem->getErrorCode(), $errorItem->getErrorCode());
}
}
}
}
public function run()
{
// Konsumsi pesan secara siklikal di thread saat ini. Kami merekomendasikan Anda menggunakan beberapa thread untuk mengonsumsi pesan secara bersamaan.
while (True) {
try {
// Konsumsi pesan dalam mode polling panjang. Konsumen mungkin menarik pesan terurut berpartisi dari beberapa partisi. Konsumen mengonsumsi pesan dari partisi yang sama dalam urutan pengiriman pesan.
// Anggaplah konsumen menarik pesan terurut berpartisi dari satu partisi. Jika broker gagal menerima ACK untuk pesan dari konsumen, broker akan mengirimkan pesan kepada konsumen lagi.
// Konsumen hanya dapat mengonsumsi batch pesan berikutnya dari partisi setelah semua pesan yang ditarik dari partisi pada batch sebelumnya diakui sebagai dikonsumsi.
// Konsumsi pesan dalam mode polling panjang. Jika tidak ada pesan dalam topik yang tersedia untuk dikonsumsi, permintaan ditangguhkan di broker selama periode waktu tertentu, yang dikenal sebagai periode polling panjang. Jika pesan tersedia untuk dikonsumsi dalam periode waktu ini, broker segera mengirimkan respons ke konsumen.
$messages = $this->consumer->consumeMessageOrderly(
3, // Jumlah maksimum pesan yang dapat dikonsumsi sekaligus. Dalam contoh ini, nilai ditentukan sebagai 3. Nilai maksimum yang dapat Anda tentukan adalah 16.
3 // Durasi periode polling panjang. Unit: detik. Dalam contoh ini, nilai ditentukan sebagai 3. Nilai maksimum yang dapat Anda tentukan adalah 30.
);
} catch (\MQ\Exception\MessageResolveException $e) {
// Jika pesan tidak dapat diuraikan karena karakter tidak valid dalam badan pesan, pengecualian ini dilemparkan.
// Pesan yang dapat diuraikan sesuai harapan.
$messages = $e->getPartialResult()->getMessages();
// Pesan yang tidak dapat diuraikan sesuai harapan.
$failMessages = $e->getPartialResult()->getFailResolveMessages();
$receiptHandles = array();
foreach ($messages as $message) {
// Logika konsumsi pesan.
$receiptHandles[] = $message->getReceiptHandle();
printf("MsgID %s\n", $message->getMessageId());
}
foreach ($failMessages as $failMessage) {
// Tangani pesan yang tidak dapat diuraikan karena karakter tidak valid dalam badan pesan.
$receiptHandles[] = $failMessage->getReceiptHandle();
printf("Gagal Menguraikan Pesan. MsgID %s\n", $failMessage->getMessageId());
}
$this->ackMessages($receiptHandles);
continue;
} catch (\Exception $e) {
if ($e instanceof MQ\Exception\MessageNotExistException) {
// Jika tidak ada pesan yang tersedia untuk dikonsumsi dalam topik, mode polling panjang terus berlaku.
printf("Tidak ada pesan, lanjutkan polling panjang!RequestId:%s\n", $e->getRequestId());
continue;
}
print_r($e->getMessage() . "\n");
sleep(3);
continue;
}
print "======>konsumsi selesai, pesan:\n";
// Logika konsumsi pesan.
$receiptHandles = array();
foreach ($messages as $message) {
$receiptHandles[] = $message->getReceiptHandle();
printf("MessageID:%s TAG:%s BODY:%s \nWaktuPublikasi:%d, WaktuKonsumsiPertama:%d, \nJumlahDikonsumsi:%d, WaktuKonsumsiBerikutnya:%d,KunciSharding:%s\n",
$message->getMessageId(), $message->getMessageTag(), $message->getMessageBody(),
$message->getPublishTime(), $message->getFirstConsumeTime(), $message->getConsumedTimes(), $message->getNextConsumeTime(),
$message->getShardingKey());
print_r($message->getProperties());
}
// Jika broker gagal menerima ACK untuk pesan dari konsumen sebelum periode waktu yang ditentukan dalam $message->getNextConsumeTime() berakhir, broker akan mengirimkan pesan kepada konsumen lagi.
// Timestamp unik ditentukan untuk handle pesan setiap kali pesan dikonsumsi.
print_r($receiptHandles);
$this->ackMessages($receiptHandles);
print "=======>pengakuan selesai\n";
}
}
}
$instance = new ConsumerTest();
$instance->run();
?>