如何将UI对接已连Redis的网络库？DASH实现Redis实时更新展示

Hey there! Let's break down your two questions one by one, with practical, actionable solutions tailored to your setup.

The core idea here is to create a clear bridge between your UI layer and the network library (which handles Redis interactions). Here are the two most common approaches depending on your needs:

Option 1: Pull data on demand (UI initiates requests) #

• First, make sure your network library exposes simple, reusable functions/methods that wrap Redis read operations. For example, if you need server messages, add a function like get_latest_server_messages(limit=10) that queries Redis and returns formatted data.
• In your UI (specifically, Dash callback functions), import these network library functions directly and call them whenever you need to refresh data. Since you're using Dash (which runs in a single process by default), this direct module import works seamlessly—no need for extra APIs or IPC.

Option 2: Push updates (UI gets notified when Redis changes) #

• Leverage Redis's Pub/Sub feature: Configure your network library to subscribe to specific Redis channels where new messages are published. When a new message comes in, the library can forward it to your UI layer.
• For Dash, since it doesn't natively support WebSockets out of the box, you can either:
  • Use dcc.Interval (we'll cover this in the next question) to periodically check for new messages from the network library, or
  • Add a lightweight WebSocket wrapper via dash-extensions (if you're open to a small dependency) for true real-time pushes.

Since you're using Dash and want to avoid Celery/standalone Flask, we'll use Dash's built-in dcc.Interval component to poll Redis (via your network library) for updates. Here's a step-by-step implementation with Server.py, Client.py, and the Dash app:

Step 1: Define roles for each file #

• Server.py/Client.py: These scripts will generate messages and write them to Redis (we'll use Redis lists to store the latest messages).
• Dash App: This will periodically pull new messages from Redis and update the UI.

Step 2: Example Code #

Server.py (simulate server sending messages to Redis)

import redis
import time
import random

# Connect to Redis (adjust host/port if your Redis is remote)
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def send_server_updates():
    while True:
        # Generate a mock server message
        message = f"Server Status: CPU {random.randint(20, 80)}% - {time.strftime('%H:%M:%S')}"
        # Push to Redis list, keep only the last 10 messages
        redis_client.lpush('server_messages', message)
        redis_client.ltrim('server_messages', 0, 9)
        time.sleep(2)  # Send update every 2 seconds

if __name__ == "__main__":
    send_server_updates()

Client.py (simulate Raspberry Pi clients sending messages)

import redis
import time
import random

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def send_client_updates():
    while True:
        # Mock client sensor data
        message = f"Client Pi: Temperature {random.uniform(20.0, 35.0):.1f}°C - {time.strftime('%H:%M:%S')}"
        redis_client.lpush('client_messages', message)
        redis_client.ltrim('client_messages', 0, 9)
        time.sleep(3)  # Send update every 3 seconds

if __name__ == "__main__":
    send_client_updates()

Dash App (real-time message display)

import dash
from dash import dcc, html, Input, Output
import redis

# Initialize Redis connection (or replace with your network library's Redis wrapper)
redis_client = redis.Redis(host='localhost', port=6379, db=0)

app = dash.Dash(__name__)

app.layout = html.Div([
    html.H1("Server & Client Real-Time Updates"),
    # Server Messages Section
    html.Div([
        html.H3("Server Status"),
        html.Div(id='server-messages', style={'padding': '10px', 'border': '1px solid #ddd'}),
    ], style={'margin': '20px 0'}),
    # Client Messages Section
    html.Div([
        html.H3("Client Pi Sensor Data"),
        html.Div(id='client-messages', style={'padding': '10px', 'border': '1px solid #ddd'}),
    ], style={'margin': '20px 0'}),
    # Interval component to trigger updates every 2 seconds
    dcc.Interval(
        id='update-interval',
        interval=2*1000,  # 2000ms = 2 seconds
        n_intervals=0
    )
])

@app.callback(
    [Output('server-messages', 'children'),
     Output('client-messages', 'children')],
    Input('update-interval', 'n_intervals')
)
def refresh_messages(_):
    # Fetch latest server messages from Redis
    server_msgs = redis_client.lrange('server_messages', 0, -1)
    # Convert bytes to strings and reverse to show oldest first
    server_display = [html.P(msg.decode('utf-8')) for msg in reversed(server_msgs)]

    # Fetch latest client messages
    client_msgs = redis_client.lrange('client_messages', 0, -1)
    client_display = [html.P(msg.decode('utf-8')) for msg in reversed(client_msgs)]

    return server_display, client_display

if __name__ == '__main__':
    app.run_server(debug=True)

Key Notes #

1. Replace Redis calls with your network library: If your network library already handles Redis connections, swap out the direct redis.Redis calls with your library's functions (e.g., from your_network_lib import get_server_messages).
2. Performance tweaks: If you need lower latency, consider using Redis Pub/Sub with dash-extensions WebSocket component (still no Celery/Flask needed). For most use cases, the dcc.Interval approach is simple and reliable.
3. Data structure flexibility: Instead of lists, you can use Redis hashes or JSON strings if you need to store structured data (e.g., redis_client.hset('server_stats', 'cpu', 50, 'memory', 75)).

内容的提问来源于stack exchange，提问作者Anhsirk Krishna