如何统计MQTT中订阅某主题的客户端数量？（基于mqtt.js与Mosca）

Great question! Since Mosca doesn’t come with a built-in way to fetch subscriber counts directly, you’ll need to track subscriptions manually on the broker side. Here’s a practical, step-by-step implementation that works smoothly with mqtt.js clients:

Step 1: Set Up Mosca Broker with Subscription Tracking #

First, we’ll spin up a Mosca broker and maintain a map to track which clients are subscribed to each topic. We’ll use Map and Set to avoid duplicate entries (since a client might re-subscribe to the same topic multiple times without creating new active subscriptions).

const mosca = require('mosca');

// Configure your Mosca broker settings
const brokerSettings = {
  port: 1883 // Default MQTT port
};

const broker = new mosca.Server(brokerSettings);

// Map to store topic -> Set of client IDs subscribed to it
const topicSubscribers = new Map();

// Triggered when the broker is ready
broker.on('ready', () => {
  console.log('Mosca broker is up and running!');
});

Step 2: Handle New Subscriptions #

Listen to Mosca’s subscribed event to add clients to our tracking map whenever they subscribe to a topic. We’ll also optionally publish the updated count to a stats topic so clients can consume it:

broker.on('subscribed', (topic, client) => {
  console.log(`Client ${client.id} subscribed to ${topic}`);
  
  // Initialize the Set for the topic if it doesn't exist
  if (!topicSubscribers.has(topic)) {
    topicSubscribers.set(topic, new Set());
  }
  
  // Add the client's ID to the topic's subscriber set
  topicSubscribers.get(topic).add(client.id);
  
  // Publish updated count to a stats topic (for client consumption)
  const currentCount = topicSubscribers.get(topic).size;
  broker.publish({
    topic: `stats/subscribers/${topic}`,
    payload: JSON.stringify({ topic, count: currentCount }),
    qos: 0,
    retain: true // Keep the latest count available for new subscribers
  });
});

Step 3: Clean Up on Unsubscription & Disconnection #

We need to update our tracking map when clients unsubscribe or disconnect entirely to keep counts accurate:

// Handle unsubscription events
broker.on('unsubscribed', (topic, client) => {
  console.log(`Client ${client.id} unsubscribed from ${topic}`);
  
  if (topicSubscribers.has(topic)) {
    const subscribers = topicSubscribers.get(topic);
    subscribers.delete(client.id);
    
    // Remove the topic entry if no subscribers are left
    if (subscribers.size === 0) {
      topicSubscribers.delete(topic);
    }
    
    // Publish updated count
    const currentCount = subscribers.size;
    broker.publish({
      topic: `stats/subscribers/${topic}`,
      payload: JSON.stringify({ topic, count: currentCount }),
      qos: 0,
      retain: true
    });
  }
});

// Handle client disconnections (clean up all their subscriptions)
broker.on('clientDisconnected', (client) => {
  console.log(`Client ${client.id} disconnected`);
  
  // Iterate through all topics to remove the client's ID
  for (const [topic, subscribers] of topicSubscribers.entries()) {
    if (subscribers.has(client.id)) {
      subscribers.delete(client.id);
      
      if (subscribers.size === 0) {
        topicSubscribers.delete(topic);
      }
      
      // Publish updated count for the affected topic
      const currentCount = subscribers.size;
      broker.publish({
        topic: `stats/subscribers/${topic}`,
        payload: JSON.stringify({ topic, count: currentCount }),
        qos: 0,
        retain: true
      });
    }
  }
});

Step 4: Get Subscriber Count Programmatically #

Add a helper function to fetch the count for any topic whenever you need it (e.g., for logging or internal broker logic):

function getSubscriberCount(targetTopic) {
  // Return 0 if the topic has no subscribers
  return topicSubscribers.get(targetTopic)?.size || 0;
}

// Example: Log the count for a specific topic every 5 seconds
setInterval(() => {
  const myTopic = 'devices/sensor/temperature';
  console.log(`Current subscribers for ${myTopic}: ${getSubscriberCount(myTopic)}`);
}, 5000);

Step 5: Access Count from mqtt.js Clients #

If you want your mqtt.js clients to receive real-time subscriber counts, have them subscribe to the stats topic we created earlier:

const mqtt = require('mqtt');

const client = mqtt.connect('mqtt://localhost:1883');
const targetTopic = 'devices/sensor/temperature';
const statsTopic = `stats/subscribers/${targetTopic}`;

client.on('connect', () => {
  client.subscribe(statsTopic, (err) => {
    if (!err) {
      console.log(`Subscribed to stats topic: ${statsTopic}`);
    }
  });
});

client.on('message', (topic, message) => {
  if (topic === statsTopic) {
    const stats = JSON.parse(message.toString());
    console.log(`Subscriber count for ${stats.topic}: ${stats.count}`);
  }
});

Important Notes #

• Wildcard Subscriptions: The above code tracks exact topic matches only. If you need to include clients subscribed to wildcard topics (e.g., devices/sensor/+ when counting devices/sensor/temperature), you’ll need extra logic to check pattern matches. You can use a lightweight library like mqtt-match to handle this.
• Duplicate Subscriptions: Using Set ensures that even if a client subscribes to the same topic multiple times, it’s only counted once (aligning with MQTT’s behavior, where duplicate subscriptions don’t create multiple active subscriptions).

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