azure-messaging-webpubsub-java

Azure Web PubSub SDK for Java

使用 Azure Web PubSub SDK for Java 构建实时 Web 应用程序。

安装

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-webpubsub</artifactId>
    <version>1.5.0</version>
</dependency>

创建客户端

使用连接字符串

import com.azure.messaging.webpubsub.WebPubSubServiceClient;
import com.azure.messaging.webpubsub.WebPubSubServiceClientBuilder;

WebPubSubServiceClient client = new WebPubSubServiceClientBuilder()
    .connectionString("<connection-string>")
    .hub("chat")
    .buildClient();

使用访问密钥

import com.azure.core.credential.AzureKeyCredential;

WebPubSubServiceClient client = new WebPubSubServiceClientBuilder()
    .credential(new AzureKeyCredential("<access-key>"))
    .endpoint("<endpoint>")
    .hub("chat")
    .buildClient();

使用 DefaultAzureCredential

import com.azure.identity.DefaultAzureCredentialBuilder;

WebPubSubServiceClient client = new WebPubSubServiceClientBuilder()
    .credential(new DefaultAzureCredentialBuilder().build())
    .endpoint("<endpoint>")
    .hub("chat")
    .buildClient();

异步客户端

import com.azure.messaging.webpubsub.WebPubSubServiceAsyncClient;

WebPubSubServiceAsyncClient asyncClient = new WebPubSubServiceClientBuilder()
    .connectionString("<connection-string>")
    .hub("chat")
    .buildAsyncClient();

核心概念

• Hub：连接的逻辑隔离单元
• Group：Hub 中的连接子集
• Connection：单个 WebSocket 客户端连接
• User：可拥有多个连接的实体

授权和投递边界

仅从经过身份验证的服务器颁发客户端令牌；用户/组/租户身份应基于可信的账户状态推导，而不是请求参数。优先使用限定组范围的角色和较短的过期时间；不受限的加入/发送角色允许访问 Hub 中的任何组。不要记录持有者 URL 或连接机密。广播/组变更/断开连接必须限定在经批准的目标范围内；服务接受请求并不代表每个客户端都已收到/处理消息。代码片段针对指定的 SDK 版本，未包含完整的身份验证/重试应用程序逻辑。

核心模式

向所有连接发送

import com.azure.messaging.webpubsub.models.WebPubSubContentType;

// Send text message
client.sendToAll("Hello everyone!", WebPubSubContentType.TEXT_PLAIN);

// Send JSON
String jsonMessage = "{\"type\": \"notification\", \"message\": \"New update!\"}";
client.sendToAll(jsonMessage, WebPubSubContentType.APPLICATION_JSON);

使用筛选条件向所有连接发送

import com.azure.core.http.rest.RequestOptions;
import com.azure.core.util.BinaryData;

BinaryData message = BinaryData.fromString("Hello filtered users!");

// Filter by userId
client.sendToAllWithResponse(
    message,
    WebPubSubContentType.TEXT_PLAIN,
    message.getLength(),
    new RequestOptions().addQueryParam("filter", "userId ne 'user1'"));

// Filter by groups
client.sendToAllWithResponse(
    message,
    WebPubSubContentType.TEXT_PLAIN,
    message.getLength(),
    new RequestOptions().addQueryParam("filter", "'GroupA' in groups and not('GroupB' in groups)"));

向组发送

// Send to all connections in a group
client.sendToGroup("java-developers", "Hello Java devs!", WebPubSubContentType.TEXT_PLAIN);

// Send JSON to group
String json = "{\"event\": \"update\", \"data\": {\"version\": \"2.0\"}}";
client.sendToGroup("subscribers", json, WebPubSubContentType.APPLICATION_JSON);

向特定连接发送

// Send to a specific connection by ID
client.sendToConnection("connectionId123", "Private message", WebPubSubContentType.TEXT_PLAIN);

向用户发送

// Send to all connections for a specific user
client.sendToUser("andy", "Hello Andy!", WebPubSubContentType.TEXT_PLAIN);

管理组

// Add connection to group
client.addConnectionToGroup("premium-users", "connectionId123");

// Remove connection from group
client.removeConnectionFromGroup("premium-users", "connectionId123");

// Add user to group (all their connections)
client.addUserToGroup("admin-group", "userId456");

// Remove user from group
client.removeUserFromGroup("admin-group", "userId456");

// Check if user is in group
boolean exists = client.userExistsInGroup("admin-group", "userId456");

管理连接

// Check if connection exists
boolean connected = client.connectionExists("connectionId123");

// Close a connection
client.closeConnection("connectionId123");

// Close with reason
client.closeConnection("connectionId123", "Session expired");

// Check if user exists (has any connections)
boolean userOnline = client.userExists("userId456");

// Close all connections for a user
client.closeUserConnections("userId456");

// Close all connections in a group
client.closeGroupConnections("inactive-group");

生成客户端访问令牌

import com.azure.messaging.webpubsub.models.GetClientAccessTokenOptions;
import com.azure.messaging.webpubsub.models.WebPubSubClientAccessToken;

// Basic token
WebPubSubClientAccessToken token = client.getClientAccessToken(
    new GetClientAccessTokenOptions());
// Return token.getUrl() only to the authenticated authorized client over TLS; never log it.

// With user ID
WebPubSubClientAccessToken userToken = client.getClientAccessToken(
    new GetClientAccessTokenOptions().setUserId("user123"));

// With roles (permissions)
WebPubSubClientAccessToken roleToken = client.getClientAccessToken(
    new GetClientAccessTokenOptions()
        .setUserId("user123")
        .addRole("webpubsub.joinLeaveGroup")
        .addRole("webpubsub.sendToGroup"));

// With groups to join on connect
WebPubSubClientAccessToken groupToken = client.getClientAccessToken(
    new GetClientAccessTokenOptions()
        .setUserId("user123")
        .addGroup("announcements")
        .addGroup("updates"));

// With custom expiration
WebPubSubClientAccessToken expToken = client.getClientAccessToken(
    new GetClientAccessTokenOptions()
        .setUserId("user123")
        .setExpiresAfter(Duration.ofHours(2)));

授予/撤销权限

import com.azure.messaging.webpubsub.models.WebPubSubPermission;

// Grant permission to send to a group
client.grantPermission(
    WebPubSubPermission.SEND_TO_GROUP,
    "connectionId123",
    new RequestOptions().addQueryParam("targetName", "chat-room"));

// Revoke permission
client.revokePermission(
    WebPubSubPermission.SEND_TO_GROUP,
    "connectionId123",
    new RequestOptions().addQueryParam("targetName", "chat-room"));

// Check permission
boolean hasPermission = client.checkPermission(
    WebPubSubPermission.SEND_TO_GROUP,
    "connectionId123",
    new RequestOptions().addQueryParam("targetName", "chat-room"));

异步操作

asyncClient.sendToAll("Async message!", WebPubSubContentType.TEXT_PLAIN)
    .subscribe(
        unused -> { },
        error -> System.err.println("Error: " + error.getMessage()),
        () -> System.out.println("Message accepted by service")
    );

asyncClient.sendToGroup("developers", "Group message", WebPubSubContentType.TEXT_PLAIN)
    .doOnSuccess(v -> System.out.println("Sent to group"))
    .doOnError(e -> System.err.println("Failed: " + e))
    .subscribe();

错误处理

import com.azure.core.exception.HttpResponseException;

try {
    client.sendToConnection("invalid-id", "test", WebPubSubContentType.TEXT_PLAIN);
} catch (HttpResponseException e) {
    System.out.println("Status: " + e.getResponse().getStatusCode());
    System.out.println("Error: " + e.getMessage());
}

环境变量

WEB_PUBSUB_CONNECTION_STRING=Endpoint=https://<resource>.webpubsub.azure.com;AccessKey=...
WEB_PUBSUB_ENDPOINT=https://<resource>.webpubsub.azure.com
WEB_PUBSUB_ACCESS_KEY=<your-access-key>

客户端角色

角色	权限
webpubsub.joinLeaveGroup	加入/离开任意组
webpubsub.sendToGroup	向任意组发送
webpubsub.joinLeaveGroup.<group>	加入/离开特定组
webpubsub.sendToGroup.<group>	向特定组发送

最佳实践

1. 使用组：将连接组织到组中，以便定向发送消息
2. 用户 ID：将连接与用户 ID 关联，以便按用户级别发送消息
3. 令牌过期时间：设置适当的令牌过期时间以确保安全
4. 角色：通过角色授予最低限度的必要权限
5. Hub 隔离：为不同的应用程序功能使用不同的 Hub
6. 连接管理：清理不活跃的连接

触发短语

• “Web PubSub Java”
• “Azure WebSocket 消息传递”
• “实时推送通知”
• “服务器发送事件”
• “聊天应用后端”
• “实时更新广播”

适用场景

当需要执行概述中描述的工作流或操作时，可以使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为针对特定环境进行验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并请求澄清。