大猩猩聊天示例與您需要的內容非常接近。該示例將從任何客戶端接收的消息廣播到所有連接的客戶端。執行以下操作以使代碼適應您的用例： 將客戶端讀取泵更改為丟棄接收到的消息，而不是將消息發送到集線器。func (c *Client) readPump() { defer func() { c.hub.unregister <- c c.conn.Close() }() c.conn.SetReadLimit(maxMessageSize) c.conn.SetReadDeadline(time.Now().Add(pongWait)) c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil }) for { if _, _, err := c.NextReader(); err != nil { break } }} 更改Kafka read循環以將消息發送到Hub.broadcast通道。for { msg, err := c.ReadMessage(xxx) if err != nil { // handle error } hub.broadcast <- msg} 刪除將客戶端發送隊列中的消息合并到單個websocke

您可以在超時后顯示消息： 發送請求時啟動計時器，該計時器完成后將顯示消息 響應到來時取消計時器。 如果響應從未出現，則無法顯示消息，因為您無法知道響應是否最終會在任何時候出現。

這將不是一個完整的答案，但我希望它能幫助你的問題。 我認為您的問題可能與您的機器的不同IP地址的混合有關。 首先，我認為Eureka公開了像host.docker.internal這樣的服務，如前所述，通過不同的docker容器引用主機的邏輯名稱，原因在這個SO問題中解釋。 基本上，看起來docker軟件正在用host.docker.internal和gateway.docker.internal的條目更新hosts文件，而Eureka可能將該別名作為正在播發的機器IP的別名。請看上述問題的公認答案。 運行Spring Boot時，底層服務器（Tomcat、Jetty、Undertow）通常會偵聽0.0.0.0地址中的連接，即所有可用的網絡接口，包括localhost。這就是困擾我的原因，因為如前所述，服務應該可以通過機器中的所有IP訪問。 無論如何，我認為你可以嘗試幾種方法來解決你的問題。 解決這個問題的最佳方法可能是將Eureka服務器和/或Eureka客戶機的hostname配置為一個公共的。 例如，您可以將服務器和客戶機配置為公開為localhost。 為此，您需要在各自的配置文件中包含以下配置屬性： eureka: instance: hostname: localhost

這是一個偉大的問題！謝謝分享 切中要害：文件明確指出 代理選項支持HTTP、HTTPS和WebSocket連接。 以及： 如果代理選項對您來說不夠靈活，您也可以： 自己配置代理 我知道。。這不是一個很好的選擇，但使用這個選項是生產也不理想。大多數情況下，您的應用程序將通過實際服務器或具有固定地址或有效域的API網關提供服務 另外請注意，如果您使用的是安全連接，則還需要管理有效的證書（或使用提供證書的管理解決方案） 簡而言之：生產和開發有不同的配置集，因此-在生產中，您將使用不同的配置，并且不會出現此錯誤。此錯誤可能僅在使用此特定配置的開發模式下生成 我希望您能滿意，因為這個答案的其余部分將更多地devOps關于生產配置，而不是開發指南

通常，（不使用Redux）它將在任何父組件將webSocketMessage作為道具傳遞給該組件的情況下處理。假設您將消息作為狀態存儲在父級中，它可能類似于。。。 import React, {useEffect, useState} from 'react';const ParentComponent = () => { const [webSocketMessage, setWebSocketMessage] = useState() useEffect(() => { if (webSocketMessage){ setTimeout(() => setWebsocketMessage(null), 30000) } }, [webSocketMessage]) return <ChildComponent webSocketMessage={webSocketMessage} />}export default ParentComponent; 同樣的想法也可以轉化為清除消息的redux操作，因為您正在使用redux存儲來保存該消息?；镜南敕ㄊ牵撠煱l送道具的任何東西（在你的例子中是Redux）也需要負責清除道具，所以你最終會。。。 const clearMessage = () => dispatch({type: 'CLEAR_ME

我假設您使用的是socket iov3，根據Nestjs gitHub存儲庫上的這個問題，這個版本暫時與Nestjs不兼容。兩種解決方案： 將套接字io版本降級為v2。 如果您不能降級，您可以按照github問題中的解釋，創建一個socket io適配器并將其添加到Main.ts中，如下所示： async function bootstrap() { const app = await NestFactory.create(AppModule); // ... app.useWebSocketAdapter(new SocketIoAdapter(app, corsOrigins));}

由于Devtools可以一次連接到多個上下文，因此可以為另一個上下文打開它，這樣軟件將是次要的，因此可以正常卸載。 打開擴展的任何可見頁面，如果沒有，則打開其manifest.jsonURL:chrome-extension://ID/manifest.json，其中ID是擴展的id 打開devtools并切換到它的Application選項卡，然后選擇左邊的Service worker。 單擊start（如果顯示）以啟動服務工作者，單擊后臺腳本名稱以在“源”面板中打開它，設置斷點，等等。 單擊stop以停止服務工作人員，可選地單擊頂部的Update，并在中間（{ skip waiting）}強制更新。 再次單擊start將觸發斷點。 啟動service worker后，您可以在左側的“源”面板（在“文件”面板中）、top-right（在threads面板中）、控制臺工具欄（上下文選擇器）中看到它的上下文。 這在一開始可能看起來很笨拙，但一旦你嘗試并掌握了它的訣竅，它就非常簡單，甚至可以擊敗在中單擊“service worker”鏈接時顯示的devtoolschrome://extensions因為這一頁a）顯示了devtools中擴展的localStorage/IndexedDB，b）提供了對服務工作者生命周期/執行的控制，c）支持軟件啟動的性能評測。 請注意，ManifestV

如果我正確理解了服務器的結果，您應該將“大小”乘以“價格”： import asyncioimport websocketsimport jsonasync def consumer() -> None: async with websockets.connect("wss://ftx.com/ws/") as websocket: await websocket.send( json.dumps( {"op": "subscribe", "channel": "trades", "market": "BTC-PERP"} ) ) totalVol = 0 async for message in websocket: message = json.loads(message) if message["type"] == "update": result = message["data"] for record in result: totalVol += record["size"] * record["price"] print("totalVol:", totalVol)asyncio.run(consumer()) 打印更“真實”的值： ...totalVol: 1420555.7282999994totalVol: 1436185.2012999994totalVo

對于angular's范圍/區域之外的任何庫提升活動，最好的辦法是將其帶到范圍/區域。 這樣，您就不需要考慮與angular生命周期事件的任何進一步同步。 我通常在服務中使用Subject或ReplaySubject，以確保所有組件都可以偵聽引發的事件。 @Injectable()export class YourService { eventType1$ = new Subject() constructor(private ngZone: NgZone ... ... socketObject.onEventType1((eventType1) => { this.ngZone.run(() => { this.eventType1$.next(eventType1) }); }) 然后，通過組件，您可以訂閱此服務的公共可觀察服務： yourComponentFunction() { this.yourService.eventType1$.subscribe((eventType1) => { // do something with eventType1 }) } 注意：請記住在組件被銷毀時完成所有訂閱，以避免常見的內存泄漏。

嗨，Rx值看起來像base64編碼字符串。使用下面的函數后，我得到了這個值，這是您所期望的嗎？ [["AUDCHF",1633525087.566,0.67279]] import base64 def base64ToString(b): return base64.b64decode(b).decode('utf-8') b = "W1siQVVEQ0hGIiwxNjMzNTI1MDg3LjU2NiwwLjY3Mjc5XV0="print(base64ToString(b))

我不是一個喜歡nodeJS的人，但我已經用其他語言多次考慮過同樣的問題。我得出的結論是，消息處理操作的速度有多慢確實很重要，因為如果它們太慢（慢于某個閾值，具體取決于msg per second的值），這可能會導致websocket連接出現瓶頸，當這個瓶頸建立起來時，可能會導致未來消息的極度延遲。 如果await和async具有與python相同的行為，如果您使用它們處理任何操作，您的處理將是異步的，這意味著它確實不會等待前一個操作被處理。 到目前為止，我想到了兩個選擇： 保持異步處理消息，但在處理消息的代碼中編寫一些附加邏輯，以管理順序混亂。例如，在繼續處理當前消息之前，請確認已處理了上一條消息。這個邏輯可能很復雜而且很慢，因為它在一個單獨的thread中運行，并且不會阻塞websocket消息。 一個接一個地同步處理消息，但只需執行一個操作即可實現極快的速度：將消息存儲在Redis中。這比將它們存儲在數據庫中要快得多，而且在大多數情況下都足夠快，不會在WS連接中造成瓶頸。然后使用單獨的進程從Redis獲取這些消息并對其進行處理。

... 改變端口和使用其他端口不是正確的答案，應該有一個解釋 這可能是答案。這取決于您的網絡管理員。 有些人似乎認為上網是危險的（這是事實），因此試圖阻止上網。當然，這是一個解決辦法，正如注意到太陽是危險的（有沒有嚴重曬傷過？），因此決定解決辦法是“消滅Sun".1 如果您的網絡管理員認為Internet危險的解決方案是阻止您個人訪問任何服務器的端口22，那么您將無法訪問github.com's端口22。其他人幾乎每個人都可以用，但你不能。 您的網絡管理員GitHub自己沒有在source—i.e.上解決這個問題，而是特別決定允許用戶通過端口443（名義上是https://協議端口）提供ssh訪問，從而繞過他們的網絡管理員。這確實對一些人有用(如果你了解互聯網協議，這種“拒絕訪問端口22”的情況可能比你最初認為的更為常見。那些不了解互聯網協議的人認為，拒絕訪問特定端口，或者除某些特定端口之外的所有端口，將是成功的。它不會，因為這是一個特殊的GitHub技巧shows.） 不幸的是，一些網絡管理員不僅會阻止某些端口，還會過濾通過未阻止端口的流量。這種過濾不應該損壞通過該端口的數據，但事實上有時確實會損壞它?，F有的技術也可以解決這一問題，2但必須在連接的兩端實現—您的計算機和GitHub的計算機都必須這樣做。 通常的解決方法是使用VPN。3考慮安裝一個。 1如果我沒記錯的話，這是我哥

在這個場景中，您充當來自受控服務器的客戶端。套接字始終是主機/服務器到客戶端/接收器。

簡而言之，您需要將await on_message(await ws.recv())更改為asyncio.create_task(on_message(await ws.recv()))。 正如您正確指出的，await對您不起作用，因為它意味著等待任務完成。盡管代碼是異步的，但在某種意義上，它是由事件循環驅動的，并且您可以并行地啟動許多這樣的任務，因此每個循環都是順序的。 await的替代方法是在后臺使用asyncio.create_task()生成作業。這將創建一個任務，該任務將以片段（兩個掛起的等待之間的每一個片段）的形式執行協同路由，并與其他活動協同路由的等效片段穿插在一起。create_task()將返回一個任務句柄，您可以（可能在某個時候應該）等待該任務完成并獲得其結果或異常。因為在您的情況下，您不關心結果，所以您甚至不需要存儲任務。

看來答案是實施“乒乓球”。這將防止防火墻或代理終止連接。 如果您ping一個websocket（客戶端或服務器），那么規范會說它必須響應（pong）。但是Javascript websocket取決于瀏覽器實現，因此最好在服務器上對所有客戶端實現30秒的ping。例如 import java.io.IOException;import java.nio.ByteBuffer;import java.util.Collections;import java.util.HashSet;import java.util.Set;import javax.websocket.OnClose;import javax.websocket.OnError;import javax.websocket.OnMessage;import javax.websocket.OnOpen;import javax.websocket.PongMessage;import javax.websocket.Session;import javax.websocket.server.ServerEndpoint;@ServerEndpoint(value = "/websockets/admin/autoreply")public class MyWebSocket { private static fina