Flutter消息队列通信插件dartzmq的使用

dartzmq 是一个基于 libzmq C++ 库实现的 Dart ZeroMQ 封装。它支持多种类型的套接字和消息传递模式，适用于构建高性能的消息队列系统。

特性

• 支持创建各种类型的套接字（如 pair, pub, sub, req, rep 等）。
• 支持发送不同类型的消息（如 List<int>, String, ZFrame 或 ZMessage）。
• 支持绑定和连接操作。
• 提供了Curve加密支持。
• 支持设置套接字选项。
• 支持接收多部分消息。
• 支持主题订阅（针对 sub 套接字）。
• 支持监控套接字状态。
• 支持异步和同步套接字。

入门指南

目前官方支持的平台为 Windows 和 Android。如果提供的 libzmq 二进制文件在你的平台上无法工作，你可能需要进行一些额外的配置。

使用示例

创建上下文

final ZContext context = ZContext();

创建异步套接字

final ZSocket socket = context.createSocket(SocketType.req);

创建同步套接字

final ZSyncSocket socket = context.createSocket(SocketType.req);

连接套接字

socket.connect("tcp://localhost:5566");

发送消息

// 发送整数列表
socket.send([1, 2, 3, 4, 5]);

// 发送字符串
socket.sendString('My Message');

// 发送帧
socket.sendFrame(ZFrame([1, 2, 3, 4, 5]));

// 发送多部分消息
var message = ZMessage();
message.add(ZFrame([1, 2, 3, 4, 5]));
message.add(ZFrame([6, 7, 8, 9, 10]));
socket.sendMessage(message, flags: ZMQ_DONTWAIT);

接收消息

接收 ZMessage

socket.messages.listen((message) {
    // 处理消息
});

接收 ZFrame

socket.frames.listen((frame) {
    // 处理帧
});

接收载荷 (Uint8List)

socket.payloads.listen((payload) {
    // 处理载荷
});

监控套接字事件

final MonitoredZSocket socket = context.createMonitoredSocket(SocketType.req);
socket.events.listen((event) {
    log('Received event ${event.event} with value ${event.value}');
});

销毁资源

socket.close();
context.stop();

示例 Demo

以下是一个完整的 Flutter 示例，展示了如何使用 dartzmq 插件：

import 'dart:async';
import 'dart:developer';

import 'package:dartzmq/dartzmq.dart';
import 'package:flutter/material.dart';

void main() {
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({Key? key}) : super(key: key);

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Flutter Demo',
      theme: ThemeData(
        primarySwatch: Colors.blue,
      ),
      home: const MyHomePage(),
    );
  }
}

class MyHomePage extends StatefulWidget {
  const MyHomePage({Key? key}) : super(key: key);

  @override
  State<MyHomePage> createState() => _MyHomePageState();
}

class _MyHomePageState extends State<MyHomePage> {
  final ZContext _context = ZContext();
  late final MonitoredZSocket _socket;
  String _receivedData = '';
  late StreamSubscription _subscription;
  int _presses = 0;

  @override
  void initState() {
    _socket = _context.createMonitoredSocket(SocketType.dealer);
    _socket.connect("tcp://localhost:5566");

    _subscription = _socket.messages.listen((message) {
      setState(() {
        _receivedData = message.toString();
      });
    });

    super.initState();
  }

  @override
  void dispose() {
    _socket.close();
    _context.stop();
    _subscription.cancel();
    super.dispose();
  }

  void _sendMessage() {
    ++_presses;
    _socket.send([_presses], flags: ZMQ_DONTWAIT);
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: const Text("dartzmq demo"),
      ),
      body: Center(
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: <Widget>[
            const Text('Press to send a message'),
            MaterialButton(
              onPressed: _sendMessage,
              color: Colors.blue,
              child: const Text('Send'),
            ),
            StreamBuilder<SocketEvent>(
              stream: _socket.events,
              builder: (context, snapshot) {
                if (snapshot.hasData) {
                  final event = snapshot.data!;
                  log('Socket event: ${event.event}, value: ${event.value}');
                  return Text('Event: ${event.event}, value: ${event.value}');
                }
                return const LinearProgressIndicator();
              },
            ),
            const Text('Received'),
            Text(_receivedData),
          ],
        ),
      ),
    );
  }
}

此示例展示了如何在 Flutter 应用中使用 dartzmq 插件来创建、连接、发送和接收消息，并监听套接字事件。请根据你的具体需求进行调整和扩展。