本文為您介紹如何使用HTTP API、SDK或eascmd訪問隊列服務。
通過API訪問隊列服務
異步推理服務部署完成后,會自動生成輸入隊列和輸出隊列(sink隊列)兩類地址,以HTTP接口為例,說明如下:
地址類型 | 地址格式 | 示例 |
輸入隊列地址 |
|
|
輸出隊列地址 |
|
|
您可以在PAI-EAS模型在線服務頁面,單擊異步推理服務的服務方式列下的調用信息,查看輸入隊列地址、輸出隊列地址和Token。
向隊列服務發送數據
使用curl
命令向輸入隊列發送一條同步請求或異步推理請求,具體代碼示例如下。
$ curl -v http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' -d '[{}]'
示例結果如下:
> POST /api/predict/qservice HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 19
< X-Eas-Queueservice-Request-Id: 4e034bnvb-e783-4272-9333-68x6a1v8dc6x
<
1033
其中:
Response Header中返回的X-Eas-Queueservice-Request-Id,為該請求對應的Request ID:4e034bnvb-e783-4272-9333-68x6a1v8dc6x,您可以通過該Request ID來查詢數據。
Response Body中返回的是當前請求在隊列中的Index:1033,您可以通過Index在當前隊列中查詢數據。
發送優先數據
在隊列服務中,普通數據按照FIFO順序進行推送,但是在很多場景中,部分數據需要被優先推送和處理。隊列服務支持數據優先推送。您可以通過增加query參數_priority_=1,向隊列服務推送優先數據。
$ curl -v http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_priority_=1 -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' -d '[{}]'
示例結果如下:
> POST /api/predict/qservice?_priority_=1 HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 19
< X-Eas-Queueservice-Request-Id: 4033eb55-e783-4922-9777-68d6a1383c76
<
1034
優先數據一旦被寫入隊列,將被優先推送給訂閱者,從而進行優先處理。
查看隊列服務詳情
如果您在向隊列服務發送請求時,增加_attrs_=true
參數,返回結果中會顯示當前隊列的詳情信息。具體代碼示例如下。
$ curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_attrs_=true
示例結果如下:
> GET /api/predict/qservice?_attrs_=true HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 320
<
{"consumers.stats.total":"0","consumers.status.total":"0","meta.header.group":"X-EAS-QueueService-Gid","meta.header.priority":"X-EAS-QueueService-Priority","meta.header.user":"X-EAS-QueueService-Uid","stream.maxPayloadBytes":"524288","meta.name":"pmml_test","meta.state":"Normal","stream.approxMaxLength":"4095","stream.firstEntry":"0","stream.lastEntry":"0","stream.length":"1"}
上述結果中返回JSON格式的詳情信息,其中關鍵字段說明如下:
字段名 | 描述 |
stream.maxPayloadBytes | 隊列中允許的每個數據項的大小上限,單位為Byte。 |
stream.approxMaxLength | 隊列中能存儲的數據項的數量上限。 |
stream.firstEntry | 隊列中第一個數據項的index。 |
stream.lastEntry | 隊列中最后一個數據項的index。 |
stream.length | 隊列中當前存儲的數據項的數量。 |
meta.state | 當前隊列的狀態。 |
您也可以在PAI-EAS模型在線服務頁面,單擊異步推理服務的名稱進入服務詳情頁面。在該頁面中,查詢隊列信息,包括隊列中當前存儲數據項數量、數據項大小上限、存儲數據項數量上限和訂閱實例數等。
查詢數據
根據條件查詢結果
當只使用一個隊列服務時,您可以通過Index或Request ID從輸入隊列中查詢數據,具體代碼示例如下。
# 通過index查詢數據。 $ curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1022 # 通過request id查詢數據。 $ curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?requestId=87633037-39a4-40bf-8405-14f8e0c31896
示例結果如下:
> GET /api/predict/qservice?_index_=1022&_auto_delete_=false HTTP/1.1 > Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com > Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****== > < HTTP/1.1 200 OK < Content-Length: 4 < Content-Type: text/plain; charset=utf-8 < [{}]
您可以配置以下參數來查詢推理結果,具體參數說明如下:
參數
類型
核心參數說明
_index_
INT
要查詢數據的起始index。默認為0,表示從隊列的初始數據項開始查詢,該index越接近被查詢數據,查詢的效率越高。
_length_
INT
要查詢的數據項的條數。默認為1,表示僅查詢一條數據項。
_auto_delete_
BOOL
是否從隊列中刪除已查詢的數據。默認為TRUE,表示查詢完成后,將查詢出的數據項自動從隊列中刪除。
_timeout_
STRING
超時時間。默認為0,表示查詢時隊列中無符合要求的數據則立即返回204狀態碼,否則等待指定時間,在超時時間內如果隊列中出現符合要求的數據,則將數據返回。示例值:1s(1秒), 1m(1分鐘)。
requestId
STRING
requestId為內建的tag,表示通過該tag來查詢數據。
說明當使用異步推理服務功能時,請求從輸入隊列返回,由EAS服務框架讀取輸出數據進行處理后將結果自動寫入到輸出隊列中,服務框架會通過requestId這個tag將輸入數據與輸出數據進行關聯,通過輸入數據的requestId即可在輸出隊列中查詢結果數據。
查詢異步推理結果
當隊列服務有與之搭配的推理服務時,推理服務會自動從輸入隊列中讀取請求數據,進行推理計算后將推理結果寫出到輸出隊列(sink)中。使用以下代碼根據Request ID(0337f7a1-a6f6-49a6-8ad7-ff2fd12b****)從輸出隊列中查詢數據。
$ curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice/sink?requestId=0337f7a1-a6f6-49a6-8ad7-ff2fd12bbe2d
示例結果如下:
> GET /api/predict/qservice/sink?requestId=0337f7a1-a6f6-49a6-8ad7-ff2fd12b**** HTTP/1.1 > Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com > Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****== > < HTTP/1.1 200 OK < Content-Length: 53 < Content-Type: text/plain; charset=utf-8 < [{"p_0":0.5224580736905329,"p_1":0.4775419263094671}]
清理數據
當您的隊列中不再需要某些數據時,可以通過API對數據進行清理。數據清理的方式主要有兩種,分別是單條數據刪除(delete)和數據截止刪除(truncate)。
刪除單條數據
# 通過index刪除數據。 $ curl -XDELETE -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1022
示例結果如下:
> GET /api/predict/qservice?_index_=1022 HTTP/1.1 > Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com > Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****== > < HTTP/1.1 200 OK < Content-Length: 4 < Content-Type: text/plain; charset=utf-8 < OK
您可以配置以下參數來查詢推理結果,具體參數說明如下:
參數
類型
核心參數說明
_index_
INT
要刪除的數據index。
批量數據刪除
# 通過index刪除數據。 $ curl -XDELETE -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1023&_trunc_=true
示例結果如下:
> GET /api/predict/qservice?_index_=1023&_trunc_=true HTTP/1.1 > Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com > Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****== > < HTTP/1.1 200 OK < Content-Length: 4 < Content-Type: text/plain; charset=utf-8 < OK
您可以配置以下參數來查詢推理結果,具體參數說明如下:
參數
類型
核心參數說明
_index_
INT
要刪除的數據截止index,低于(不包含)這個index的數據將被刪除。
_trunc_
BOOL
在批量刪除時必須為true,否則將轉換為單條刪除。
隊列服務訂閱推送
在異步推理場景中,除了上述的阻塞查詢,您還可以通過訂閱的方式來獲取推理結果。隊列服務提供了訂閱(watch)接口,客戶端可以通過該接口來獲取推理結果。隊列服務根據當前推理服務實例配置的并發數(worker_threads)來控制訂閱的窗口(Window)大小,當隊列中被寫入新數據時,隊列服務會自動將數據推送給正在訂閱的客戶端。
該功能在SDK中基于WebSocket協議封裝了客戶端實現QueueClient,通過長連接的方式建立推送鏈路。下面以一個典型的視頻、語音流處理場景為例,介紹如何通過Python SDK中的QueueClient來訂閱隊列中的數據。
推理服務不是必須的,您也可以通過SDK在自定義的服務中訂閱隊列服務的輸入隊列,輸出結果也可以選擇寫入到第三方的消息隊列中或其它目標存儲中(比如輸出圖片到OSS)。
安裝EAS Python SDK。
pip install eas_prediction --user
通過QueueClient的
put()
方法向輸入隊列中發送數據,并使用watch()
方法從輸出隊列中訂閱數據。在實際使用場景中,發送數據和訂閱數據可以由不同的線程處理,本示例中發送數據和訂閱數據在同一線程中完成,先put數據,后watch結果。#!/usr/bin/env python from eas_prediction import QueueClient # 創建輸入隊列對象,用于寫入輸入數據。 input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice') # 如果需要自定義user和group,可以分別通過uid和gid進行指定,示例如下: # input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice', uid='your_user_id', gid='your_group_id') input_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==') input_queue.init() # 創建輸出隊列對象,用于訂閱讀取輸出結果數據。 sink_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice/sink') sink_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==') sink_queue.init() # 各輸入隊列中推送10個數據項。 for x in range(10): index, request_id = input_queue.put('[{}]') print(index, request_id) # 查看輸入隊列的詳情。 attrs = input_queue.attributes() print(attrs) # 從輸出隊列中watch數據,窗口為5。 i = 0 watcher = sink_queue.watch(0, 5, auto_commit=False) for x in watcher.run(): print(x.data.decode('utf-8')) # 每次收到一個請求數據后處理完成后手動commit。 sink_queue.commit(x.index) i += 1 if i == 10: break # 關閉已經打開的watcher對象,每個客戶端實例只允許存在一個watcher對象,若watcher對象不關閉,再次運行時會報錯。 watcher.close()
通過eascmd訪問隊列服務
eascmd已經封裝好了完整的隊列服務API,您可以使用eascmd stream子命令快速操作、調試隊列服務。
下載eascmd
請確保eascmd版本大于2.6.0,您可以參考文檔下載并認證客戶端來下載、更新、配置eascmd命令行客戶端。
配置eascmd訪問隊列服務
通過easmd stream config命令配置需要訪問的隊列服務,示例如下:
eascmd stream config --url=http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice --token=YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
上述配置完成后,eascmd默認使用的group_id和user_id分別是default_group和default_user,關于group和user的概念,請您參考文檔隊列服務訂閱推送。如果您需要使用其他的group_id或者user_id,您可以通過--group參數和--user參數進行指定。stream config中的所有參數,都可以在其他讀寫命令的執行中進行覆蓋。
查詢隊列詳情
使用info命令查看隊列信息。示例如下:
eascmd stream info
示例結果如下:
[OK] Attributes:
consumers.list.[0] : Id: imageasync.imageasync-35d72370-5f576f7c8d-2mdb4, Index: 0, Pending: 0, Status: Running, Idle: 19.997s, Window: 5, Slots: 5, AutoCommit: false
consumers.stats.total : 1
consumers.status.total : 1
groups.list.[0] : Id: imageasync, Index: 0, Pending: 0, Delivered: 1, Consumers: 1
meta.header.group : X-EAS-QueueService-Gid
meta.header.priority : X-EAS-QueueService-Priority
meta.header.user : X-EAS-QueueService-Uid
meta.maxPayloadBytes : 8192
meta.name : imageasync-queue-38895e88
meta.state : Normal
stream.approxMaxLength : 230399
stream.firstEntry : 0
stream.lastEntry : 0
stream.length : 0
返回中的參數的具體說明請您參見通過API訪問隊列服務。info命令不僅可以讓您觀察隊列的屬性,也可以讓您測試與隊列服務的連通性。
向隊列中發送數據
使用put命令向隊列中發送數據,示例如下:
eascmd stream put -d "10s"
示例結果如下:
[OK] 1
[INFO] Put data done.
Total time cost: 401.892141ms
Total size: 3.00 B
Total: 1, success: 1, failed: 0
您也可以通過-f參數選擇將文件中的數據全部發送到隊列,如下所示:
eascmdm stream put -f test.data
示例結果如下:
[INFO] Opening data file: test.data
[OK] 2
[OK] 3
[OK] 4
[OK] 5
[OK] 6
[OK] 7
[OK] 8
[OK] 9
[OK] 10
[OK] 11
[OK] 12
[OK] 13
....
此時您可以通過info命令觀察隊列狀態。
查詢隊列中的數據
使用get命令從隊列中查詢數據,如下所示:
eascmd stream get -l10 --timeout=3s
示例結果如下:
[OK] [0 - 1] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=e47b76e2-2648-40fe-9197-a268015cbd1f ts@source=1685802680575] data1
[OK] [1 - 2] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=51d13952-6ba3-4d52-b548-e58837675c7a ts@source=1685807531686] data2
[OK] [2 - 3] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f ts@source=1685807531701] OK
[OK] [3 - 4] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=2a5645b6-9ee5-4026-bdee-fab31e435934 ts@source=1685807531715] data4
[OK] [4 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@source=1685807531730] data5
...
以收到的第一條數據介紹整體的輸出格式:
第一列 [0 - 1] 表示收到的0號數據index為1。
第二列
tags[Header:Content-Type=text/plain; charset= ...]
表示該數據帶有的標簽(tag)。其中:以Header開頭的是您輸入數據時使用的HTTP請求頭。
requestId為內建的自動生成的請求ID。
ts@source表示輸入隊列在收到您請求時的unix時間戳,與之對應的還有ts@sink為輸出隊列在收到數據時的時間戳。
最后一列為您輸入的數據。
如果您有與之搭配的推理服務實例,在向隊列輸入數據后可能會被推理服務實例消費掉。此時,您需要在輸出隊列中查詢數據,具體做法是在命令中增加-k參數。
您也可以通過--tags參數增加查詢條件,比如當需要通過requestId來進行查詢時,可以使用以下命令:
eascmd stream get --tags requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f
示例結果如下:
[OK] [0 - 3] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f ts@source=1685807531701] OK
刪除隊列中的數據
通過delete和trunc命令進行單條數據刪除和批量數據刪除,參考下述命令:
單條刪除:
eascmd stream delete 3
在確認之后,示例結果如下:
Deleting index(es):
3 [y/N]y
[OK] deleted
批量刪除:
eascmd stream trunc 4
在確認之后,示例結果如下:
trunc stream from index: 4 [y/N]y
[OK] truncated
訂閱隊列
通過watch命令訂閱隊列服務,參考下述命令:
eascmd stream watch
示例結果如下:
[INFO] Start to watch: index: 0, indexOnly: false, autoCommit: false, window: 10
I0604 09:20:45.211243 66197 queue.go:532] watch via websocket
[OK] [0 - 4] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=2a5645b6-9ee5-4026-bdee-fab31e435934 ts@sink=1685807531718 ts@source=1685807531715] data4
commit: 4 ? [Y/n]
當您輸入Y之后就可以將該數據進行commit,會得到新的數據:
[OK] [1 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
commit: 5 ? [Y/n]
當您輸入n之后可以確認是否進行negative commit:
[OK] [1 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
commit: 5 ? [Y/n]n
negative: 5 ? [Y/n]y
關于commit及negative commit的說明,請參見Commit與Negative。
如果您使用--auto-commit選項,將在server端自動commit數據:
eascmd stream watch --auto-commit
示例結果如下:
[INFO] Start to watch: index: 0, indexOnly: false, autoCommit: true, window: 10
I0604 09:30:08.554542 66408 queue.go:532] watch via websocket
[OK] [0 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
[OK] [1 - 6] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=5825dd3e-a5e2-4754-a946-96e068d643c8 ts@sink=1685807531771 ts@source=1685807531768] data6
[OK] [2 - 7] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=e7edf9b8-de78-41a0-8d9c-0a4aaf7dcaaf ts@sink=1685807531786 ts@source=1685807531783] data7
[OK] [3 - 8] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=3ddc3481-934a-4408-8d08-11c2c2248ef6 ts@sink=1685807531801 ts@source=1685807531798] data8
[OK] [4 - 9] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=561da95d-b99a-4710-bb82-9402baa21f36 ts@sink=1685807531816 ts@source=1685807531812] data9
....
其它選項與命令
上文介紹了eascmd stream的主要命令與選項,eascmd stream還有其它擴展功能,您可以通過eascmd stream help命令獲取更多詳細幫助。