如何在Xamarin Forms Android后台Service中集成RabbitMQ客户端并解决收不到消息问题

应用场景说明 #

• 应用包含News（新闻页）和Announcements（公告页），需要展示RabbitMQ服务器推送的消息
• Android平台实现了后台Service，在用户登录完成后启动
• 共享代码与Android平台通信使用Xamarin.MessagingCenter
• 测试环境：RabbitMQ服务运行在虚拟机，应用运行在模拟器，控制台测试应用可以正常连接RabbitMQ并接收消息，只有应用内的客户端无法收到任何推送

已实现代码 #

Android端 DataTransferTaskService 实现 #

[Service]
class DataTransferTaskService : Service
{
    static User user = new User { Groups = new List<string>() { "Test","Test2" } };
    CancellationTokenSource _cts;
    public override IBinder OnBind(Intent intent)
    {
        return null;
    }

    public override StartCommandResult OnStartCommand(Intent intent, StartCommandFlags flags, int startId)
    {
        _cts = new CancellationTokenSource();
        PS();
        return StartCommandResult.NotSticky;
    }

    void PS()
    {
        var factory = new ConnectionFactory() { HostName = "10.0.0.3", UserName = "test", Password = "test", Port = 5672 };
        var connection = factory.CreateConnection();
        var channel = connection.CreateModel();
        
        channel.ExchangeDeclare(exchange: "Kastner", type: ExchangeType.Direct);
        var queueName = channel.QueueDeclare().QueueName;

        foreach (string g in user.Groups)
        {
            channel.QueueBind(queue: queueName,
                              exchange: "Kastner",
                              routingKey: g);
        }

        var consumer = new EventingBasicConsumer(channel);
        System.Diagnostics.Debug.WriteLine("事件触发前");
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            var obj = JObject.Parse(message);
            News n;
            Announcement a;
            System.Diagnostics.Debug.WriteLine("if判断前");
            if (obj.Properties().Select(p => p.Name).FirstOrDefault() == "NewsId")
            {
                n = JsonConvert.DeserializeObject<News>(message);
                MessagingCenter.Send<object, News>(this, "NewsMessage", n);
            }
            else
            {
                a = JsonConvert.DeserializeObject<Announcement>(message);
                MessagingCenter.Send<object, Announcement>(this, "AnnouncementMessage", a);
            }
        };
        channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
    }
}

Android端 MainActivity 实现 #

[Activity(Label = "DA_MessageBrokerApp", Icon = "@mipmap/icon", Theme = "@style/MainTheme", MainLauncher = true, ConfigurationChanges = ConfigChanges.ScreenSize | ConfigChanges.Orientation | ConfigChanges.UiMode | ConfigChanges.ScreenLayout | ConfigChanges.SmallestScreenSize )]
public class MainActivity : global::Xamarin.Forms.Platform.Android.FormsAppCompatActivity
{
    protected override void OnCreate(Bundle savedInstanceState)
    {
        TabLayoutResource = Resource.Layout.Tabbar;
        ToolbarResource = Resource.Layout.Toolbar;

        base.OnCreate(savedInstanceState);
        FFImageLoading.Forms.Platform.CachedImageRenderer.Init(enableFastRenderer: true);
        Xamarin.Essentials.Platform.Init(this, savedInstanceState);
        global::Xamarin.Forms.Forms.Init(this, savedInstanceState);
        CrossMediaManager.Current.Init(this);

        LoadApplication(new App());
        WireUpDataTransferTask();
    }

    public override void OnRequestPermissionsResult(int requestCode, string[] permissions, [GeneratedEnum] Android.Content.PM.Permission[] grantResults)
    {
        Xamarin.Essentials.Platform.OnRequestPermissionsResult(requestCode, permissions, grantResults);
        base.OnRequestPermissionsResult(requestCode, permissions, grantResults);
    }

    void WireUpDataTransferTask()
    {
        MessagingCenter.Subscribe<NewsViewModel>(this, "StartDataTransferMessage", (sender) =>
        {
            var intent = new Intent(this, typeof(DataTransferTaskService));
            StartService(intent);
        });
    }
}

共享代码层 NewsViewModel 实现 #

public class NewsViewModel : BaseViewModel
{
    static User user = new User { Groups = new List<string>() { "Test","Test2" } };
    private News _selectedItem;

    public ObservableCollection<News> Items { get; }
    public Command LoadItemsCommand { get; }
    public Command AddItemCommand { get; }
    public Command<News> ItemTapped { get; }

    public NewsViewModel()
    {
        Title = "News";
        Items = new ObservableCollection<News>();
        MessagingCenter.Subscribe<object, News>(this, "NewsMessage", async (sender, arg) =>
        {
            await Task.Run(() =>DataStoreNews.AddItemAsync(arg));
            await ExecuteLoadItemsCommand();
        });
        MessagingCenter.Send<NewsViewModel>(this, "StartDataTransferMessage");

        LoadItemsCommand = new Command(async () => await ExecuteLoadItemsCommand());
        ItemTapped = new Command<News>(OnItemSelected);
    }

    void AddNewItem(News news)
    {
        if (!Items.Contains(news))
            Items.Add(news);
    }

    async Task ExecuteLoadItemsCommand()
    {
        IsBusy = true;
        try
        {
            Items.Clear();
            var items = await DataStoreNews.GetItemsAsync(true);
            foreach (var item in items)
            {
                Items.Add(item);
            }
        }
        catch (Exception ex)
        {
            Debug.WriteLine(ex);
        }
        finally
        {
            IsBusy = false;
        }
    }

    public void OnAppearing()
    {
        IsBusy = true;
        SelectedItem = null;
    }

    public News SelectedItem
    {
        get => _selectedItem;
        set
        {
            SetProperty(ref _selectedItem, value);
            OnPropertyChanged(value);
        }
    }

    async void OnItemSelected(News item)
    {
        if (item == null)
            return;
        await Shell.Current.GoToAsync($"{nameof(NewsDetailPage)}?{nameof(NewsDetailViewModel.NewsId)}={item.NewsId}");
    }
}

共享代码层 BaseViewModel 实现 #

public class BaseViewModel : INotifyPropertyChanged
{
    public IDataStore<News> DataStoreNews => DependencyService.Get<IDataStore<News>>();
    public IDataStore<Announcement> DataStoreMessage => DependencyService.Get<IDataStore<Announcement>>();

    bool isBusy = false;
    public bool IsBusy
    {
        get { return isBusy; }
        set { SetProperty(ref isBusy, value); }
    }

    string title = string.Empty;
    public string Title
    {
        get { return title; }
        set { SetProperty(ref title, value); }
    }

    protected bool SetProperty<T>(ref T backingStore, T value,
        [CallerMemberName] string propertyName = "",
        Action onChanged = null)
    {
        if (EqualityComparer<T>.Default.Equals(backingStore, value))
            return false;
        backingStore = value;
        onChanged?.Invoke();
        OnPropertyChanged(propertyName);
        return true;
    }

    #region INotifyPropertyChanged
    public event PropertyChangedEventHandler PropertyChanged;
    protected void OnPropertyChanged([CallerMemberName] string propertyName = "")
    {
        var changed = PropertyChanged;
        if (changed == null)
            return;
        changed.Invoke(this, new PropertyChangedEventArgs(propertyName));
    }
    #endregion
}

核心问题 #

PS() 方法中集成的RabbitMQ客户端消息接收事件始终不触发，控制台测试RabbitMQ连接、消息推送均正常。

补充说明 #

已移除DataTransferTaskService类中connection和channel的using语句，避免触发异常，但收不到消息的问题仍然存在。

问题原因与解决方案 #

1. 连接对象被GC回收 #

当前实现中connection、channel、consumer都是PS()方法内的局部变量，方法执行完成后这些对象就失去了全局引用，会被Android垃圾回收机制回收，RabbitMQ连接直接断开。即使移除了using语句，也无法解决这个问题，因为GC会回收无引用的对象。
解决：将连接、通道、消费者对象提升为Service类的成员变量，保持全局引用：

// 类成员变量定义
IConnection _connection;
IModel _channel;
EventingBasicConsumer _consumer;

void PS()
{
    // 原有逻辑不变，不再创建局部变量
    var factory = new ConnectionFactory() { HostName = "10.0.0.3", UserName = "test", Password = "test", Port = 5672 };
    _connection = factory.CreateConnection();
    _channel = _connection.CreateModel();
    // 其余队列声明、绑定、消费逻辑与原代码一致
}

2. Android 9+ 明文流量限制 #

Android 9及以上版本默认禁止应用发送未加密的TCP流量，RabbitMQ默认5672端口使用明文通信，会被系统直接拦截，且不会抛出明显异常。
解决：在AndroidManifest.xml中开启明文流量权限，同时确认已声明INTERNET权限：

<uses-permission android:name="android.permission.INTERNET" />
<application android:usesCleartextTraffic="true" ...>
</application>

3. 主线程网络IO阻塞 #

CreateConnection()、CreateModel()都是同步网络IO操作，直接在主线程执行会触发NetworkOnMainThreadException，当前代码没有捕获该异常，会导致Service启动后直接崩溃退出，不会执行后续消费逻辑。
解决：将PS()方法的执行放到后台线程：

public override StartCommandResult OnStartCommand(Intent intent, StartCommandFlags flags, int startId)
{
    _cts = new CancellationTokenSource();
    // 后台线程执行，避免主线程阻塞
    Task.Run(() => PS(), _cts.Token);
    return StartCommandResult.NotSticky;
}

4. 临时队列生命周期问题 #

当前使用无参QueueDeclare()会创建仅当前连接有效的随机临时队列，连接断开后队列会自动删除，服务器推送的消息会直接丢弃。
解决：替换为持久化命名队列，确保应用离线时消息可留存：

// 替换原有的随机队列声明
var queueName = "app_message_queue_" + user.Id;
_channel.QueueDeclare(queue: queueName,
    durable: true,
    exclusive: false,
    autoDelete: false,
    arguments: null);

5. 无自动重连机制 #

默认RabbitMQ客户端没有开启网络断开后的自动恢复功能，模拟器网络波动、虚拟机网络抖动都会导致连接断开且不会重连。
解决：在ConnectionFactory中开启自动恢复：

var factory = new ConnectionFactory() 
{ 
    HostName = "10.0.0.3", 
    UserName = "test", 
    Password = "test", 
    Port = 5672,
    AutomaticRecoveryEnabled = true,
    NetworkRecoveryInterval = TimeSpan.FromSeconds(10)
};

6. MessagingCenter 订阅时序问题 #

当前NewsViewModel在构造函数中发送StartDataTransferMessage消息，但此时Service的订阅还未完成绑定，会导致Service启动失败。
解决：将Service启动逻辑移到用户登录完成后的统一入口，不要通过ViewModel构造函数触发。

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