导入包:
https://github.com/gaufung/rocketmq-client-dotnet/tree/master
using org.apache.rocketmq.client.consumer.listener;
using org.apache.rocketmq.client.producer;
using System;
using System.Text;
using java.util;
using System.Windows.Forms;
using org.apache.rocketmq.client.consumer;
using org.apache.rocketmq.common.consumer;
using System.Threading;
using System.Collections.Generic;
namespace wf_RMQ3
{
public partial class Form1 : Form
{
public Form1()
{
InitializeComponent();
}
private void button1_Click(object sender, EventArgs e)
{
try
{
DefaultMQProducer p = new DefaultMQProducer("PG01");
p.setNamesrvAddr("172.20.168.210:9876");
p.setInstanceName("PER01");
//p.setProducerGroup("myproducer");
p.start();
var data = Encoding.UTF8.GetBytes(textBox1.Text.ToString());
org.apache.rocketmq.common.message.Message m = new org.apache.rocketmq.common.message.Message("TP01", data);
p.send(m);
p.shutdown();
textBox1.Text = "";
}
catch (Exception ex)
{
throw ex;
}
}
private void button2_Click(object sender, EventArgs e)
{
try
{
//DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
////consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//consumer.setNamesrvAddr("172.20.168.210:9876");
//consumer.subscribe("jinwei01", "*");
//consumer.registerMessageListener(new TestListener());
//consumer.start();
//启动推送型消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CG01");
consumer.setNamesrvAddr("172.20.168.210:9876");
consumer.subscribe("TP01", "*");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.setConsumeTimestamp("20170422221800");
consumer.setInstanceName("CER01");
//consumer.setConsumerGroup("myconsumer01");
//设置消费者端口,官方没有该功能。适用端口有安全限制的服务器
//if (port > 0) consumer.setClientPort(port);
//注册推送事件
consumer.registerMessageListener(new ChainwayMessageListener());
//启动消费者
consumer.start();
}
catch (Exception ex)
{
throw ex;
}
}
}
public class ChainwayMessageListener : MessageListenerConcurrently
{
public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext ccc)
{
for (int i = 0; i < list.size(); i++)
{
var msg = list.get(i) as org.apache.rocketmq.common.message.Message;
byte[] body = msg.getBody();
var str = Encoding.UTF8.GetString(body);
MessageBox.Show(str);
Console.Write(str);
continue;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
}