1.首先我从官方下载了rabbitmq-server
当然公司还是以java开发为主,因此下载的版本是基于linux下面的,不过我再这里下载一个ubuntu版本的,因为我开发的机子就是ubuntu.
下载好后,安装很快,ubuntu下直接dpkg -i rabbitmq-server_3.5.0-1_all.deb
当然其他linux系统可以下载tar.gz包解压,但是还需要安装erlang.解压后可以执行以下命令就可以启动rabbitmq了
解压的目录sbin下找到rabbitmq-server运行即可,当然这里不需要详细说安装了,可以参考官方文档进行安装.
当然,为了我们能可视化查看rabbitmq-server的运行情况和满足其他开发需求我们可以启用插件
rabbitmq-plugins enables,这样我们可以打开http://127.0.0.1:15672/
使用guest,guest登陆后可以看到http管理界面.
2.理解rabbitmq的工作原理
第一步安装好服务后,接下来我们就要进行生产消息和消费消息了.
但是要开发必须要理解rabbitmq的工作原理.当然如果我长篇大论记下来估计要记录几十页,因此如果英语比较好的可以先看看官方文档.
http://www.rabbitmq.com/getstarted.html
当然如果英语实在是渣的话,不妨参考下面的链接:
http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/
看完这些文档之后,估计会对rabbitmq理解的差不多了.
当然网上也有很多人写的博客,都可以参照的.
3.java客户端开发
当然按照上述文档我们可以开发出自己需要的客户端,但是这里我主要介绍如何使用spring-amqp来进行开发.
首先我们需要引入spring-amqp包,为了能查看调试信息,我们也引入log4j
org.springframework.amqp spring-rabbit ${spring-rabbit.version} log4j log4j 1.2.17
引入后就要进行开发了,当然spring-amqp如何配置这里不想长篇大论,因此这里只介绍两种发送方式
异步发送消息
其实这种方式是很容易的,大多数网上文章都是基于配置文件的,但是如果我们使用代码来创建,那么更加会理解spring-amqp的设计原理的,能使用代码创建也就更加能使用配置文件了
CachingConnectionFactory factory=new CachingConnectionFactory();//创建链接工厂,使用带缓存功能的工厂可以基于Channel创建多个Channel来处理//设置rabbitmq-server链接地址factory.setAddresses("localhost:5672");//设置登陆用户名,可以在rabbit-server http页面创建用户factory.setUsername("guest");//设置登陆密码(在本机上可以不用设置)factory.setPassword("guest");//启动channel的个数并缓存factory.setChannelCacheSize(channelSize);//创建admin进行队列绑定,exchange绑定RabbitAdmin admin=new RabbitAdmin(factory);//创建名字为qingting-queue的队列,durable是否是持久化消息到服务端Queue queue=new Queue("qingting-queue",durable);//此时会在rabbitmq-servers上创建一个queueadmin.declareQueue(queue);//创建exchange并进行订阅到服务端,这里创建的是topic类型的exchange,也就是路由key可以是正则表达式Exchange exchange=new TopicExchange("qingting-exchange",durable,autoDelete);//rabbitmq-server上会创建一个名称交exchangeName的exchange,同时类型为topic,而且特征有D表示durableadmin.declareExchange(exchange);/*** 使用routeKey将队列绑定到exchange上面,这个时候,rabbitmq-server上的queue和exchange会有一个通过routeKey的映射关系*/admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("qingting.*").noargs());//创建一个Template来进行发消息,也可以收消息等操作template = new RabbitTemplate(factory);//这里讨论发消息,我们通过exchange来进行发消息,同时设置routeKey,这时候因为是topic类型的exchange,通过比对设置的routeKey跟上面绑定的bindRouteKey做正则匹配,如果匹配上了就会把消息推送到queueName的这个队列template.setExchange("qingting-exchange");template.setRoutingKey("qingting.route.key");//设置将消息转换为json类型,当然自己可以设置其他的转换器或者自己实现template.setMessageConverter(new Jackson2JsonMessageConverter());
有了上面配置后,那么我们可以进行发送同步消息了
template.convertAndSend();
因为上述设置了消息转换器,因此最好使用改方法,当然还有template.send()方法其实使用该方法确实结果Jackson2JsonMessageConverter将消息内容体处理成json并转换为字节数组,
同时还会将消息的属性配置中conentType设置为application/json 并且head中设置一些消息内容的对象类型等信息,之所以这样做是因为,消费者接收消息,也可以设置json转换器直接将拿接收的消息属性中head的值将消息体内容进行转换成对象,可以在rabbitmq管理页面看到发送的消息会附带如下信息
Exchange: qingting-exchange //上述template设置的exchangeRouting Key: qingting.route.key //template设置的routeKeyProperties headers: __KeyTypeId__:java.lang.Object //转json附带的信息,方便转换 __TypeId__:java.util.HashMap __ContentTypeId__:java.lang.Object content_encoding:UTF-8 content_type:application/json Payload {"username":"81a40c08-5a01-46fa-93c3-981fcccb3c8c","index":0} 因此可以看到http页面上在绑定的队列会有一条消息到绑定的队列,因为这时候没有消费者订阅队列,我们设置的是持久化durable,因此会一一直留在server中