依赖
dart_amqp: ^0.3.1 #rabbitMq接收发送消息
工具封装
import 'package:dart_amqp/dart_amqp.dart'; /// 封装 RabbitMQ 的服务类 class RabbitMQService { late ConnectionSettings _settings; // RabbitMQ 连接设置 late Client _client; // RabbitMQ 客户端 late Channel _channel; // RabbitMQ 通道 /// 初始化 RabbitMQ 客户端 /// [host] 主机地址 /// [port] 端口号 /// [username] 用户名 /// [password] 密码 Future<void> initialize({ required String host, required int port, required String username, required String password, }) async { print('初始化RabbitMQ客户端…'); _settings = ConnectionSettings( host: host, port: port, authProvider: PlainAuthenticator(username, password), ); _client = Client(settings: _settings); print('连接RabbitMQ…'); try { _channel = await _client.channel(); print('通道创建成功。'); } catch (e) { print('创建通道失败:: $e'); rethrow; } } /// 发送消息到指定的交换机 /// [exchange] 交换机名称 /// [routingKey] 路由键 /// [message] 要发送的消息内容 Future<void> publishMessage( String exchange, String routingKey, String message) async { // 获取指定的交换机 final exchangeInstance = await _channel.exchange(exchange, ExchangeType.DIRECT); // 发送消息到交换机 exchangeInstance.publish(message, routingKey); } /// 直接发送消息到指定队列 /// [queueName] 队列名称 /// [message] 要发送的消息内容 Future<void> publishMessageToQueue(String queueName, String message) async { // 获取指定的队列 final queue = await _channel.queue(queueName, durable: true); // 发送消息到队列 queue.publish(message); } /// 从指定队列中消费消息 /// [queueName] 队列名称 /// [onMessage] 消息处理回调函数 Future<void> consumeMessages( String queueName, void Function(AmqpMessage) onMessage) async { // 获取指定的队列 final queue = await _channel.queue(queueName, durable: true); // 消费队列中的消息 final consumer = await queue.consume(); // 监听队列中的消息 consumer.listen((AmqpMessage message) async { print('监听队列中的消息: $message'); try { // 处理消息 onMessage(message); // 确认消息已处理 // message.ack(); //rabbitmq自动确定 可以不加 } catch (e) { print('处理消息时出错: $e'); // 确保在处理失败时记录异常 } }); } /// 关闭 RabbitMQ 连接和通道 Future<void> close() async { // 关闭通道 await _channel.close(); // 关闭客户端 await _client.close(); } }
使用
//建立通道 init() async { await rabbitMQService.initialize( host: '39.13.110', // 替换为你的 RabbitMQ 主机地址 port: 5672, // 默认 RabbitMQ 端口号 username: 'ain', // 替换为你的 RabbitMQ 用户名 password: 'hachjhb', // 替换为你的 RabbitMQ 密码 ); } await rabbitMQService.publishMessageToQueue( "队列", //队列的名称 'Hello RabbitMQ! 峰的 $a' // 要发送的消息内容 ); Future<void> xiaofei() async { // 订阅消息,从指定队列中获取消息 await rabbitMQService.consumeMessages('队列', (AmqpMessage message) { // 打印收到的消息内容 print('到的消息内容: ${message.payloadAsString}'); setState(() { word = message.payloadAsString; }); }); }
标签:String,队列,await,RabbitMQ,rabbitmq,使用,message,flutter,消息 From: https://www.cnblogs.com/xbinbin/p/18327383