《基于 Kafka + Flink + ES 實現危急值處理措施推薦和范圍校準》

📢 大家好，我是【戰神劉玉棟】，有10多年的研發經驗，致力于前后端技術棧的知識沉淀和傳播。 💗
🌻 近期剛轉戰 CSDN，會嚴格把控文章質量，絕不濫竽充數，歡迎多多交流。👍

文章目錄

寫在前面的話
- 背景技術
- 發明目的
- 具體方案
- - 一、前置環境準備
  - 二、核心服務實現
  - 三、與用戶門戶的交互實現
  - 四、開發危急值分析模塊
  - 五、開發危急值醫務管理模塊
- 危急值發送流程
- **危急值處理流程**
- 方案特征
- 總結陳詞

寫在前面的話

本篇文章分享一下博主所在公司的危急值處理措施推薦和范圍校準的實現方案。
主要是基于 Kafka + Flink + Elasticsearch 實現，由于涉及安全問題，內容以方案介紹為主，有需要探討的可以留言。
好，讓我們開始。

背景技術

危急值是指當這種檢驗、檢查結果出現時，表明患者可能正處于生命危險的邊緣狀態，臨床醫生需要及時得到檢驗、檢查信息，迅速給予患者有效的干預措施或治療，就可能挽救患者生命。危急值信息可供臨床醫生對生命處于危險邊緣狀態患者采取及時、有效的治療，避免病人因意外發生，出現嚴重后果，失去最佳搶救機會。
危急值報告制度的制定與實施，能有效增強醫技工作人員的主動性和責任心，提高醫技工作人員的理論水平，增強醫技人員主動參與臨床診斷的服務意識，促進臨床、醫技科室之間的有效溝通與合作。危急值管理是醫院管理的重要組成部分，危急值的快速甄別、確認、發布、及時接收以及對該流程監控、分析等是系統信息化管理的目標與方向。
現目前大多危急值管理系統存在如下問題：
1、各類危急值檢驗項目按照固定的參考范圍進行危急值判定，不支持動態調整項目范圍的上限和下限，往往只是簡單粗暴的進行數值比對，缺乏科學的范圍判定和校準方式，造成了不符合臨床實際的“假”危急值頻繁通知，對臨床醫護人員工作產生較大影響；
2、醫護人員針對危急值做出的干預措施，只是單純的填寫措施并反饋醫技部門，沒有與病程和護理記錄產生交互，處理過程也沒有形成記憶和管理，針對相同的情形的工作，往往需要花費重復工作和時間，也容易產生偏差；
3、危急值涉及環節較廣，沒有統一的流程化管理，容易產生環節缺失，未形成完整閉環，同時，整個過程缺少環節監控、日志跟蹤、和異常處理方案，也缺少全院統計匯總頁面，無法提供整體改善醫院危急值的方案；

發明目的

本專利發明的目的是基于 Kafka + Flink + Elasticsearch 等技術，在危急值全流程管理過程中，實現一種科學校準危急值判定范圍、以及對危急值處理措施進行智能推薦的方案，以解決現目前危急值流程管理中存在的諸如危急值判定標準不準確、處理措施數據未利用、全院危急值優化機制不完善等問題，進而優化危急值流程、提升危急值處理效率、形成完整閉環追蹤，最終建設完善的危急值全流程管理系統。
1、建設“假”危急值處理方案，防止提醒過于頻繁，不斷校準危急值項目的合理范圍；
2、針對日常危急值的處理措施進行存儲，形成記憶，在危急值到來時，可以給予醫護人員多樣化提示，充當填寫助手的作用；
3、基于消息中心事件驅動機制，建設完整危急值處理流程，盡可能包含臨床業務場景的各個環節，由數據中心負責完整危急值信息存儲，提供閉環展示和數據查詢接口；收集各科室的危急值閉環數據，生成定向指標數據，方便定期追蹤、分析、評價危急值指標，督導各科室發現并完善自身危急值處理，提升全院危急值處理效率；

具體方案

本方案是基于 Kafka + Flink + Elasticsearch 實現危急值范圍校準和措施推薦，具體技術方案實現如下。

一、前置環境準備

1、部署 Kafka 環境，程序引入 Kafka 相關依賴，并進行相關配置與功能集成，定義“危急值發送”和“危急值反饋”事件，同時配置事件的消息入參格式與XSD校驗文本，這兩個事件將作為 Kafka 的兩個主題 Topic，其中，Kafka 用于充當消息中間件，負責提供生產者和消費者的協作模式；
2、部署 Elasticsearch 環境，程序引入 Elasticsearch 相關依賴，并進行相關配置與功能集成，Elasticsearch 定義若干索引結構，將作為危急值原始數據、關聯數據、運算結果等內容的存儲，并利用 Elasticsearch 特性，進行統計分析；
3、部署 Flink 環境，程序引入 Flink 相關依賴，并進行相關配置與功能集成，Flink 充當呈上啟下的銜接角色，一方面用于消費 Kafka 投遞的主題消息，另一方面，通過相關 API，將數據運算后，輸出存儲到 Elasticsearch 當中；

二、核心服務實現

1、提供對外的消息生產者接口
開發消息中心生產者接口，并對外部系統開放，該接口可以用于“危急值發送”和“危急值反饋”這兩個場景。
主要邏輯是，針對消息入參進行合理性校驗、解析和處理，再通過調用 Kafka API 進行消息發送，利用生產者單例去完成消息發送。
發送的主題 Topic 為“危急值發送”或“危急值反饋”。

2、利用 Flink 消費 Kafka
利用 Flink 的** **Flink Source API，添加 Kafka 作為數據來源，并訂閱“危急值發送”和“危急值反饋”這兩個 Topic。
針對拉取到的消息，添加消息消費處理的代碼塊。

3、利用 Flink 加工流數據
3.1、危急值發送流程
針對拉取到的 Kafka 的“危急值發送”Topic 主題數據，進行相應加工處理。
1）利用正則表達式提取消息入參中危急值核心屬性內容，包含但不限于危急值ID、報告ID、患者ID、就診ID等，識別出各關鍵屬性的code和value，并組裝為 Map 結構；
2）利用上述關鍵信息，從 Oracle 中提取危急值業務關聯的報告信息、患者信息、就診信息，以及上述內容的歷史信息，從 Elasticsearch 中提取危急值區間分布信息、危急值處理措施分布等信息，將這些內容組裝，用于輔助分析；
3）利用 Flink Transform API 對 Map 數據進行綜合加工處理，得到相關結果；
4）危急值發送過程中，關于范圍校準和措施推薦的相關運算如下：
a、獲取該項目危急值的基本信息，判斷該危急值是否符合當前危急值的上限和下限；
b、獲取該項目危急值的區間分布情況，判斷該危急值所屬的區間分布，做出更新，并將結果組裝；
c、獲取該項目危急值的歷史處理措施分布情況，并通過運算得出，按不同維度的不同措施出現的頻次排列，再將結果組裝；
d、獲取該項目危急值歷史出現時，其他同樣異常的項目，并通過運算得出，這些項目和當前危急值項目之間存在的聯動關系；
e、獲取該項目危急值對應項目的歷史值，做出趨勢分析，并將結果組裝；
f、獲取該項目危急值的其他關聯和擴展信息，用于輔助分析，并將結果組裝；
g、將原始危急值數據存入Elasticsearch，再將第3步的所有運算結果進行組裝，進入下一環節，充當填寫助手的作用。

3.2、危急值處理流程
針對拉取到的 Kafka 的“危急值處理”Topic 主題數據，進行相應加工處理。
1）利用正則表達式提取消息入參中危急值處理的屬性內容，包含但不限于危急值ID、處理方式、處理措施、處理人等，識別出各關鍵屬性的code和value，并組裝為 Map 結構；
2）同危急值發送，利用上述關鍵信息，從 Oracle 和 Elasticsearch 中提取關聯信息，用于輔助分析
3）利用 Flink Transform API 對 Map 數據進行綜合加工處理，得到相關結果；
4）危急值處理過程中，關于范圍校準和措施推薦的相關運算如下：
a、若醫生針對該危急值給予了正常的干預措施，則代表危急值觸發范圍的可信度增加，首先更新該項目危急值出現的頻次記錄信息；然后，增加該項目組值此時可以更新危急值范圍區間數據，代表某項目出現危急值的范圍區間更加精準，本區間如果本次危急值屬于原有出現危急值的區間內，則增加區間出現次數，若本次危急值超出原有出現危急值的區間，則新增區間數據，擴大區間范圍，并進行次數記錄；最后，將醫生的處理措施和危急值數值，關聯存儲到措施記憶索引中，該索引記錄包含但不限于如下內容：不同項目所使用的處理措施，屬于哪個區間，歷史觸發數值包含哪些，關聯患者的當前和歷史的報告、就診、危急值等信息。
b、若醫生針對該危急值給予了異常處理，例如點擊了反饋疑問按鈕，則代表危急值觸發范圍的可信度降低，首先插入危急值關鍵信息到反饋疑問索引中；然后，往危急值范圍區間索引中也插入相關異常數據；最后，也將更新處理措施索引，反饋疑問也屬于處理措施的一個環節；這些疑問內容都將在提供相應統計分析頁面，人工進行最后裁定，范圍是否變更；

4、利用 Flink 輸出到 Elasticsearch
利用 Flink Elasticsearch API，添加 ElasticsearchSink 作為結果輸出，將上一步計算得出的結果，按照不同維度存儲到ES的不同索引結構。
包含但不下于如下索引：危急值原始數據索引、危急值擴展數據索引、危急值區間頻次分布索引、危急值處理措施分布索引等。

三、與用戶門戶的交互實現

3.1、危急值發送流程
經過核心服務的數據處理，可以調用用戶統一門戶的后端接口，再利用 WebSocket 完成前后端消息推送，或由核心服務直接集成 WebSocket 負責與門戶前端交互，最終在用戶門戶前端展示出危急值霸屏彈窗界面。
霸屏彈窗上醫生除了看到危急值對應基本信息、報告信息、患者信息外，還可以填寫干預措施提交，或點擊反饋疑問按鈕。
霸屏彈窗上將展示如下填報助手信息：
a、該項目危急值的各項處理措施出現的頻次，醫生可以快速點擊復用；
b、該項目不同觸發區間出現危急值的頻次，作為醫生確認危急值的參考依據；
c、該項目的歷史趨勢對比分析圖，以及該項目出現危急值時，其他產生危急值時，同樣異常的項目信息出現的頻次；
d、其他歷史參考信息，例如就診歷史、報告歷史、危急值歷史等；
3.2、危急值處理流程
醫生針對危急值霸屏彈窗做出處理，將調用統一門戶后端接口，觸發 Kafka 的“危急值處理”Topic 主題數據，進入核心服務的危急值處理環節。
醫生有兩種處理模式，可以填寫干預措施提交，或點擊反饋疑問按鈕，兩種方式都可以結束處理流程。

四、開發危急值分析模塊

6.1、設置定時服務，利用聚合函數對ES的數據進行二次加工處理，處理結果繼續存儲在新的索引空間下。
6.2、開發前端BI界面，將加工前后的危急值指標數據進行展示，并給出分析提示。
1）針對參考范圍給出分析判定結果，允許人工最后確認校準范圍是否改動。
2）針對處理措施給出推薦分析，針對不同處理措施，按照使用次數、處理路徑、對應值范圍，以及歷史項目趨勢、關聯其他并發項目、歷史診病信息等內容，給出各項統計、指引、分析。

五、開發危急值醫務管理模塊

1）定義危急值評價指標：醫務科應定義危急值評價指標，如：處理率%，平均處理時常h，及時處理率/24小時處理率%，患者六小時復診率%，危急值處理總數等；
2）危急值統計：醫務科需要定期對各科室危急值管理制度執行情況進行督導檢查、追蹤、分析，定期評價危急值報告、處置的及時性。在醫務門戶的危急值組件，匯聚展示各科室的危急值數據，按科室和醫生兩個維度，展示各類指標的排行榜和詳情數據，并支持導出報表一覽展示數據，定期比對全院不同時間點的指標對比，制定全院危急值階段改善計劃；
3）危急值反饋：醫務科需要根據臨床實際情況對危急值項目及危急值進行更新調整，將科室危急值管理納入科室醫療質量考核，在醫務人員門戶開發危急值反饋組件，統一收集、分析和處理這些反饋。

危急值發送流程

流程：LIS - 數據中心 - 生產者 - Kafka Source - Flink - 加工處理 - Elasticsearch
1、醫技人員發現危急值情況時，檢查（驗）者首先要確認檢查儀器、設備和檢驗過程是否正常，核查標本是否有錯，操作是否正確，儀器傳輸是否有誤，在確認臨床及檢查（驗）過程各環節無異常的情況下，及時復查（影像科室可根據實際情況決定是否需要復查），如兩次復查結果相同，才可以將檢查（驗）結果發出。
2、檢查（驗）系統發出危急值后，將通過院內集成平臺發起對數據中心危急值發送接口的調用，首先將進行危急值的存儲，接著調用本方案中的消息中心生產者接口，投遞“危急值發送”Topic 至 Kafka；
3、本方案的核心服務，將利用 Flink 訂閱 Kafka 的危急值發送 Topic，并利用 Flink Transform API 對接收到的數據進行加工處理，形成需要的數據，再利用 Flink Elasticsearch API，添加 ElasticsearchSink 將結果輸出到 Elasticsearch 相關索引中；
4、經過核心服務的數據處理，可以調用用戶統一門戶的后端接口，再利用 WebSocket 完成前后端消息推送，或由核心服務直接集成 WebSocket 負責與門戶前端交互。最終在用戶門戶前端展示出危急值霸屏彈窗界面；
5、至此，發送流程結束。

危急值處理流程

流程：門戶 - 數據中心 - 生產者 - Kafka Source - Flink - 加工處理 - Elasticsearch
1、醫生用戶在日常使用門戶系統過程中，若收到危急值發送通知，則會以霸屏彈窗的形式進行展示；
2、醫生根據患者的危急值信息，以及報告信息、就診信息等內容進行判斷，若確認符合危急值范疇，則填寫相應干預措施，觸發數據中心的危急值處理邏輯；
3、數據中心首先更新危急值信息，然后將通過院內集成平臺發起對檢查（驗）系統危急值發送接口的調用，緊接著調用本方案中的消息中心生產者接口，投遞“危急值處理”Topic 至 Kafka；
4、本方案的核心服務，將利用 Flink 訂閱 Kafka 的危急值處理 Topic，并利用 Flink Transform API 對接收到的數據進行加工處理，形成需要的數據，再利用 Flink Elasticsearch API，添加 ElasticsearchSink 將結果輸出到 Elasticsearch 的相關索引中；
5、若第2步，醫生判斷該危急值屬于誤報，則點擊“反饋疑問”按鈕，將調用“醫務管理模塊”的錯誤問題上報接口，以便后續分析；

方案特征

1、基于 Kafka + Flink 組合實現，利用了大數據流式引擎技術的優勢，針對危急值的發送和處理場景，實現高可靠、高效實時、高擴展性的數據加工，最終實現危急值范圍校準和措施推薦的目的；
2、利用 Elasticsearch 存儲多樣化的指標運算結果，再利用 ES 的聚合函數功能對結果二次分析處理，整體方案擴展性和可重用性都獲得較大的提升；
3、將消息中心事件驅動機制應用于危急值場景，為危急值閉環流程的關鍵節點建立消息事件，通過動態訂閱的方式為事件指定訂閱服務，流程清晰可插拔。以消息中心為樞紐建立完整危急值處理流程，盡可能覆蓋實際業務場景的各個環節，提升業務覆蓋面和人員參與度；