钉钉单聊机器人 Stream 模式接入实现

注册长连接信息成功后，客户端将获取长连的身份标识 ticket 和钉钉开放平台的地址信息，通过此信息客户端和钉钉服务端建立一条 WebSocket 连接，握手请求的路径和参数信息如下所示：

GET /connect?ticket=${ticket} HTTP/1.1
Host: wss-open-connection.dingtalk.com
Upgrade: websocket

以上示例中 Host、Path 信息仅用于示例展示，请用步骤一中返回的 endpoint 信息作为 Host 和 Path 构建 WebSocket 请求。

钉钉服务端收到 WebSocket 握手信息后会通过 ticket 校验客户端身份信息，校验成功后会返回正确的握手信息。

至此，已经完成了推送订阅通道建立，可以实时接收到订阅的消息列表。

Java 实现

• 运行环境：JDK 1.8 及以上。
• 安装 Java SDK：添加依赖项到工程的 pom.xml 文件或下载对应的 jar 包，最新的 SDK 版本可以在 这里 查看和下载。

<dependency>
    <groupId>com.dingtalk.open</groupId>
    <artifactId>dingtalk-stream</artifactId>
    <version>{sdk-version}</version>
</dependency>

钉钉 Stream 流的构建

创建 IM 消息的监听：

@Configuration
public class DingTalkStreamClientConfiguration {
    @Value("${app.appKey}")
    private String clientId;

    @Value("${app.appSecret}")
    private String clientSecret;

    /**
     * 配置 OpenDingTalkClient 客户端并配置初始化方法 (start)
     *
     * @param chatBotCallbackListener
     * @return
     * @throws Exception
     */
    @Bean(initMethod = "start")
    public OpenDingTalkClient configureStreamClient(@Autowired ChatBotCallbackListener chatBotCallbackListener) throws Exception {
        // init stream client
        return OpenDingTalkStreamClientBuilder.custom()
                // 配置应用的身份信息，企业内部应用分别为 appKey 和 appSecret，三方应用为 suiteKey 和 suiteSecret.
                .credential(new AuthClientCredential(clientId, clientSecret))
                // 注册机器人回调
                .registerCallbackListener(DingTalkStreamTopics.BOT_MESSAGE_TOPIC, chatBotCallbackListener)
                .build();
    }
}

机器人消息回调

/**
 * 机器人消息回调
 *
 * @author zeymo
 */
@Slf4j
@Component
public class ChatBotCallbackListener implements OpenDingTalkCallbackListener<ChatbotMessage, JSONObject> {
    private RobotPrivateMessagesService robotPrivateMessagesService;

    @Autowired
    public ChatBotCallbackListener(RobotPrivateMessagesService robotPrivateMessagesService) {
        this.robotPrivateMessagesService = robotPrivateMessagesService;
    }

    /**
     * https://open.dingtalk.com/document/orgapp/the-application-robot-in-the-enterprise-sends-group-chat-messages
     *
     * @param message
     * @return
     */
    @Override
    public JSONObject execute(ChatbotMessage message) {
        try {
            MessageContent text = message.getText();
            if (text != null) {
                String msg = text.getContent();
                log.info("receive bot message from user={}, msg={}", message.getSenderId(), msg);
                String openConversationId = message.getConversationId();
                try {
                    // 发送机器人消息
                    robotPrivateMessagesService.send(openConversationId, "hello");
                } catch (Exception e) {
                    log.error("send group message by robot error:" + e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            log.error("receive group message by robot error:" + e.getMessage(), e);
        }
        return new JSONObject();
    }
}

以上代码实现了这些能力：

1. 通过命令行参数读取 Client ID 和 Client Secret 选项
2. 通过 Client ID 和 Client Secret 创建 Stream Client
3. 在 Stream Client 中注册机器人消息回调方法，实现消息接收能力
4. 在消息回调方法中，简单 echo 机器人消息回去，实现消息发送 (回复) 能力

在 IDE 中运行 BotEchoMarkdownApplication.java 中 main 函数，当看到这样的日志输出时候表示运行成功 [DingTalk] connection is established, connectionId=...

配置事件订阅

前提条件

1. 拥有所在钉钉组织开发者后台的 开发者权限。
2. 拥有所在钉钉组织的 企业内部应用。
3. 已经完成 开发 Stream 模式推送服务端（推荐） 流程。

操作步骤

1. 登录 开发者后台，单击目标应用，进入应用详情页。
2. 单击 开发配置 > 事件订阅，选择 Stream 模式推送。
3. 服务端开发完成后，单击 已完成接入，验证连接通道。
4. 单击 保存。保存完成后，事件订阅列表才会展示。

常见问题与解决方案

1. 重新发送导致重复事件

原因是网络延迟不可控，如果因为互联网的正常抖动导致推送延迟，触发超时重新推送的话，就会出现重复事件。因此无论是否正确的 ACK 了，都需要考虑到收到重复事件的可能性。

备注：机器人消息当前是 fire-and-forget 模式，不会因为网络超时而重复推送。

2. 多个实例监听同一个机器人

只会有一个消费。确保只能被服务器消费。

解决方案：使用 Conditional 控制 IP 地址来阻止 Stream 流连接。

ConditionalOnIp.java

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Conditional(RobotCondition.class)
public @interface ConditionalOnIp {
    /**
     * 允许的 IP 地址列表，支持通配符和 CIDR 表示法
     * 例如："192.168.1.*", "10.0.0.0/24", "127.0.0.1"
     */
    String[] allowed() default {};

    /**
     * 禁止的 IP 地址列表
     */
    String[] denied() default {};

    /**
     * 配置属性名称，从配置文件中读取 IP 列表
     */
    String value() default "";

    /**
     * 当无法获取 IP 时是否匹配（默认 false）
     */
    boolean matchIfMissing() default false;
}

RobotCondition.java

public class RobotCondition implements Condition {
    private static final String LOCAL_IP = getLocalIp();

    @Override
    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
        Map<String, Object> annotationAttributes = metadata.getAnnotationAttributes(ConditionalOnIp.class.getName());
        if (annotationAttributes == null) {
            return false;
        }
        String[] allowedIPs = (String[]) annotationAttributes.get("allowed");
        String[] deniedIPs = (String[]) annotationAttributes.get("denied");
        String configProperty = (String[]) annotationAttributes.get("value");
        boolean matchIfMissing = (boolean) annotationAttributes.get("matchIfMissing");
        
        if (StringUtils.hasText(configProperty)) {
            Environment env = context.getEnvironment();
            String configValue = env.getProperty(configProperty);
            if (configValue != null) {
                allowedIPs = configValue.split(",");
            } else {
                return matchIfMissing;
            }
        }
        String currentIP = LOCAL_IP;
        if (currentIP == null) {
            return matchIfMissing;
        }
        // 检查禁止列表
        if (deniedIPs.length > 0 && isIPInList(currentIP, deniedIPs)) {
            return false;
        }
        // 检查允许列表
        if (allowedIPs.length > 0) {
            return isIPInList(currentIP, allowedIPs);
        }
        // 如果没有设置允许列表，且不在禁止
        return true;
    }

    /**
     * 判断 IP 是否在列表中
     */
    private boolean isIPInList(String ip, String[] ipList) {
        for (String ipPattern : ipList) {
            if (matchesIP(ip, ipPattern.trim())) {
                return true;
            }
        }
        return false;
    }

    /**
     * IP 匹配逻辑，支持通配符和 CIDR
     */
    private boolean matchesIP(String ip, String pattern) {
        // 精确匹配
        if (ip.equals(pattern)) {
            return true;
        }
        // 通配符匹配：192.168.1.*
        if (pattern.contains("*")) {
            String regex = pattern.replace(".", "\\.").replace("*", ".*");
            return Pattern.matches(regex, ip);
        }
        // CIDR 表示法匹配：192.168.1.0/24
        if (pattern.contains("/")) {
            try {
                return isInCIDRRange(ip, pattern);
            } catch (Exception e) {
                return false;
            }
        }
        // IP 范围匹配：192.168.1.1-192.168.1.100
        if (pattern.contains("-")) {
            try {
                return isInIPRange(ip, pattern);
            } catch (Exception e) {
                return false;
            }
        }
        return false;
    }

    /**
     * CIDR 范围匹配
     */
    private boolean isInCIDRRange(String ip, String cidr) {
        String[] parts = cidr.split("/");
        String network = parts[0];
        int prefixLength = Integer.parseInt(parts[1]);
        long ipLong = ipToLong(ip);
        long networkLong = ipToLong(network);
        long mask = (0xFFFFFFFFL) << (32 - prefixLength);
        return (ipLong & mask) == (networkLong & mask);
    }

    /**
     * IP 范围匹配
     */
    private boolean isInIPRange(String ip, String range) {
        String[] ips = range.split("-");
        long start = ipToLong(ips[0].trim());
        long end = ipToLong(ips[1].trim());
        long ipLong = ipToLong(ip);
        return ipLong >= start && ipLong <= end;
    }

    /**
     * IP 转 long
     */
    private long ipToLong(String ip) {
        String[] octets = ip.split("\\.");
        long result = 0;
        for (int i = 0; i < 4; i++) {
            result |= Long.parseLong(octets[i]) << (24 - (8 * i));
        }
        return result;
    }

    private static String getLocalIp() {
        // 优先获取回环地址
        try {
            Enumeration<NetworkInterface> netInterfaces = NetworkInterface.getNetworkInterfaces();
            while (netInterfaces.hasMoreElements()) {
                NetworkInterface iface = netInterfaces.nextElement();
                if (iface.isLoopback() || !iface.isUp()) {
                    continue;
                }
                Enumeration<InetAddress> addresses = iface.getInetAddresses();
                while (addresses.hasMoreElements()) {
                    InetAddress addr = addresses.nextElement();
                    if (!addr.isLoopbackAddress() && iface.getDisplayName().indexOf("Virtual") == -1) {
                        return addr.getHostAddress();
                    }
                }
            }
            // 如果没有找到，返回会回环地址
            return InetAddress.getLocalHost().getHostAddress();
        } catch (SocketException e) {
            return "127.0.0.1";
        } catch (UnknownHostException e) {
            return "127.0.0.1";
        }
    }
}

接下来就可以设置配置文件或者设置 allowed 属性来控制 IP 了。

3. 响应消息处理是否支持负载均衡方式处理？

补充问题：启动多个程序订阅相同事件进行数据处理，担心将来推送数据量大时可能出现无法及时响应的情况；可能需要考虑方案提供给客户端侧进行参考。

支持的。如果事件量较大的话，可以采用多进程，或者单进程下多 Stream Client 实例方式，建立多个 Stream 通道，也即多个 WebSocket 长连接。钉钉服务端每次推送消息时候，通过随机策略选取一个 Stream 通道推送。如果需要支持更多的负载均衡策略，可以通过技术支持提交反馈。

4. 注册连接凭证中的 localIp 是否可以标记多个 IP？

补充问题：或增加自定义的客户端标识参数【可选】

可以支持多个 IP，采用英文逗号分隔。