activeMq笔记

时间:2022-06-07 09:35:30

对比

http://jm.taobao.org/2016/04/01/kafka-vs-rabbitmq-vs-rocketmq-message-send-performance/

安装

下载地址:http://activemq.apache.org/download.html

安装教程: http://gerrard-ok.iteye.com/blog/1766203

解压缩:

运行: ./activemq start

.Net使用

教程:http://www.cnblogs.com/madyina/p/4121458.html#3249312

下载:http://activemq.apache.org/nms/activemq-downloads.html

还有一个下载地址比较全: http://archive.apache.org/dist/activemq/apache-nms/

控制台:http://ip:8161/

用户名: admin, 密码:admin

发送连接: tcp://192.168.16.23:61616?wireFormat.maxInactivityDuration=0

接收连接:failover:(tcp://192.168.16.23:61616?wireFormat.maxInactivityDuration=0&maxInactivityDurationInitalDelay=30000&connection.AsyncSend=true)

Send方法:

    public class FileUploadedMq : IActiveMq
{
private static string ConnString;
public static ConnectionFactory Factory { get; private set; } //private static IMessageProducer MsgProducer; public static event Action<FileUploadedModel> Recved; static FileUploadedMq()
{
ConnString = dbo.GetDbConnString(MqTypeEnum.FileUpload.ToString()); Factory = new ConnectionFactory(ConnString);
} public void Send(FileUploadedModel Msg)
{
try
{
using (IConnection connection = Factory.CreateConnection())
{
//通过连接创建Session会话
using (ISession session = connection.CreateSession())
{
//通过会话创建生产者,方法里面new出来的是MQ中的Queue
var MsgProducer = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("FileUploaded"));
//创建一个发送的消息对象
var TxtMsg = MsgProducer.CreateTextMessage();
TxtMsg.Properties.SetString("filter", "FileUploaded"); //给这个对象赋实际的消息
TxtMsg.Text = Msg.ToJson();
//设置消息对象的属性,这个很重要哦,是Queue的过滤条件,也是P2P消息的唯一指定属性
//生产者把消息发送出去,几个枚举参数MsgDeliveryMode是否长链,MsgPriority消息优先级别,发送最小单位,当然还有其他重载
MsgProducer.Send(TxtMsg, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
}
}
}
catch (Exception ex)
{
InfoTypeEnum.Error.LogTo(ex.Message);
} }
}

注册Receive:

    public class ActiveMq
{
public static FileUploadedMq FileUploaded = new FileUploadedMq();
public static FileResolvedMq FileResolved = new FileResolvedMq();
public static FileFastDfsSavedMq FileFastDfsSaved = new FileFastDfsSavedMq();
// public static EmailUploadMq EmailUpload=new EmailUploadMq(); public static bool RegisteFileUploaded(Action<FileUploadedModel> msgRecv, Action<FileUploadedModel, Exception> ErrorFunc = null)
{
return RegisteMq(MqTypeEnum.FileUpload, o =>
{
if (msgRecv == null) return;
msgRecv(o.FromJson<FileUploadedModel>());
}, (a, b) =>
{
if (ErrorFunc == null) return;
ErrorFunc(a.FromJson<FileUploadedModel>(), b);
});
} public static bool RegisteFileResolved(Action<FileResolvedModel> msgRecv, Action<FileResolvedModel, Exception> ErrorFunc = null)
{
return RegisteMq(MqTypeEnum.FileResolved, o =>
{
if (msgRecv == null) return;
msgRecv(o.FromJson<FileResolvedModel>());
}, (a, b) =>
{
if (ErrorFunc == null) return;
ErrorFunc(a.FromJson<FileResolvedModel>(), b);
});
} public static bool RegisteFileFastDfsSaved(Action<FileFastDfsSavedModel> msgRecv, Action<FileFastDfsSavedModel, Exception> ErrorFunc = null)
{
return RegisteMq(MqTypeEnum.FileFastDfsSaved, o =>
{
if (msgRecv == null) return;
msgRecv(o.FromJson<FileFastDfsSavedModel>());
}, (a, b) =>
{
if (ErrorFunc == null) return;
ErrorFunc(a.FromJson<FileFastDfsSavedModel>(), b);
});
} private static bool RegisteMq(MqTypeEnum Name, Action<string> msgRecv, Action<string, Exception> ErrorFunc)
{
if (msgRecv == null) return false; var QueueName = Name.ToString();
try
{
var factory = FileResolvedMq.Factory; //通过工厂构建连接
IConnection connection = factory.CreateConnection();
//这个是连接的客户端名称标识
//connection.ClientId = Environment.MachineName + "FileResolvedMqListener"; //启动连接,监听的话要主动启动连接
connection.Start();
//通过连接创建一个会话
ISession session = connection.CreateSession();
//通过会话创建一个消费者,这里就是Queue这种会话类型的监听参数设置
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue(QueueName));//, "filter='FileResolved'"); session.DeleteDestination(QueueName); //注册监听事件
consumer.Listener += message =>
{
var txt = (ITextMessage)message;
if (txt == null || txt.Text.HasValue() == false) { return; } var model = txt.Text; try
{
msgRecv.Invoke(model);
}
catch (Exception e)
{
InfoTypeEnum.Error.LogTo(QueueName + ":在处理过程出现错误!" + connection.ClientId, model.ToJson(), e.Message); if (ErrorFunc != null)
{
try
{
ErrorFunc(model, e);
}
catch { }
}
}
};
return true;
}
catch (Exception ex)
{
InfoTypeEnum.Error.LogTo(QueueName + ":" + ex.Message);
return false;
}
}
}

问题

1. 查看消息显示: Error!  Exception occurred while processing this request, check the log for more information!

原因: 安装了 JRE8 , 改到 JRE7!

参考:http://bbs.csdn.net/topics/390811825

安装老版本:http://www.java.com/zh_CN/download/faq/other_jreversions.xml

activeMq笔记的更多相关文章

  1. ActiveMQ笔记&lpar;7&rpar;:如何清理无效的延时消息?

    ActiveMQ的延时消息是一个让人又爱又恨的功能,具体使用可参考上篇ActiveMQ笔记(6):消息延时投递,在很多需要消息延时投递的业务场景十分有用,但是也有一个缺陷,在一些大访问量的场景,如果瞬 ...

  2. ActiveMQ笔记&lpar;6&rpar;:消息延时投递

    在开发业务系统时,某些业务场景需要消息定时发送或延时发送(类似:飞信的短信定时发送需求),这时候就需要用到activemq的消息延时投递,详细的文档可参考官网说明,本文只介绍二种常用的用法: 注:本文 ...

  3. ActiveMQ笔记&lpar;5&rpar;:JMX监控

    系统上线运行后,及时监控报警是很必要的手段,对于ActiveMQ而言,主要监控的指标有:MQ本身的健康状况.每个队列的生产者数量.消费者数量.队列的当前消息数等. ActiveMQ支持JMX监控,使用 ...

  4. ActiveMQ笔记&lpar;4&rpar;:搭建Broker集群&lpar;cluster&rpar;

    上一篇介绍了基于Networks of Borkers的2节点HA方案,这一篇继续来折腾Networks of Brokers,当应用规模日渐增长时,2节点的broker可能仍然抗不住访问压力,这时候 ...

  5. ActiveMQ笔记&lpar;3&rpar;:基于Networks of Brokers的HA方案

    上一篇介绍了基于ZK的ActiveMQ HA方案,虽然理解起来比较容易,但是有二个不足: 1)  占用的节点数过多,1个zk集群至少3个节点,1个activemq集群也至少得3个节点,但其实正常运行时 ...

  6. ActiveMQ笔记&lpar;2&rpar;:基于ZooKeeper的HA方案

    activemq官网给出了3种master/slave的HA方案,详见:http://activemq.apache.org/masterslave.html,基于共享文件目录,db,zookeepe ...

  7. ActiveMQ笔记&lpar;1&rpar;:编译、安装、示例代码

    一.编译 虽然ActiveMQ提供了发布版本,但是建议同学们自己下载源代码编译,以后万一有坑,还可以尝试自己改改源码. 1.1 https://github.com/apache/activemq/r ...

  8. ActiveMQ笔记——技术点汇总

    目录 · Introduction to ActiveMQ · Installing ActiveMQ · Message-oriented middleware · JMS specificatio ...

  9. ActiveMq笔记3-AMQ高可用性理论

    单点的ActiveMQ作为企业应用无法满足高可用和集群的需求,所以ActiveMQ提供了master-slave.broker cluster等多种部署方式,但通过分析多种部署方式之后我认为需要将两种 ...

随机推荐

  1. JAVA集合类汇总

    一.集合与数组 数组(可以存储基本数据类型)是用来存现对象的一种容器,但是数组的长度固定,不适合在对象数量未知的情况下使用. 集合(只能存储对象,对象类型可以不一样)的长度可变,可在多数情况下使用. ...

  2. queryString&lpar;正则表达式版本&rpar;

    获取所有query string function queryStringAll(s) { var reg = /(?:^|&)([^&]+)=([^&]+)(?=&| ...

  3. Server Develop (九) Simple Web Server

    Simple Web Server web服务器hello world!-----简单的socket通信实现. HTTP HTTP是Web浏览器与Web服务器之间通信的标准协议,HTTP指明了客户端如 ...

  4. 【Path Sum】cpp

    题目: Given a binary tree and a sum, determine if the tree has a root-to-leaf path such that adding up ...

  5. HTML5 例子学习 HT 图形组件

    HTML5 例子学习 HT 图形组件 HT 是啥:Everything you need to create cutting-edge 2D and 3D visualization. 这口号是当年心 ...

  6. 畅通工程--hdu1232&lpar;并查集&rpar;

    畅通工程 Time Limit: 4000/2000 MS (Java/Others)    Memory Limit: 65536/32768 K (Java/Others)Total Submis ...

  7. 批量安装XP补丁的命令

    方法一: 新建一个bat文件,把这个文件和所有补丁文件放在同一个目录下,双击运行. BAT代码如下 @echo offfor %%i in (*.exe) do %%i /passive /nores ...

  8. 千万不要随意在网上下载ojdbcjar包来使用,ORA-01461错误解决

    我在登录项目时,点击某一按钮提示ORA-01461: 仅能绑定要插入 LONG 列的 LONG 值,但是项目在我的同事那里可以完好运行.最后百度 发现问题所在: 数据库与客户端的JDBC驱动不匹配. ...

  9. egret贝塞尔曲线运动

    class MtwGame { public constructor() { } private static _instance: MtwGame; public static get Instan ...

  10. C 排序 解题报告

    C 排序 题意 给一个\(1\sim n(n\le 10^6)\)排列,求这个排列用冒泡排序从小到大排序的第\(cnt\)步的状态.这里步的定义为,比较一次算一步. 贴个我看的不是很懂的题解,嗯不是很 ...