前言
消息中間件目前已經在很多大型的項目上得到了運用,我們常見的有 RabbitMq, activitymq,kafka,rocketmq,其中rocketmq是阿里自己在kafka的基礎上用java寫的一個消息中間件。在我們使用的支付寶等應用中都有大量的使用。
附幾種常見的mq對比圖如下:
環境搭建
我們都知道大部分情況下都是在linux系統下部署服務。但是本篇的環境是在Windows下。
1.安裝Erlang
rabbitMQ是一個在AMQP協議標准基礎上完整的,可服用的企業消息系統。它遵循Mozilla Public License開源協議,采用 Erlang 實現的工業級的消息隊列(MQ)服務器,Rabbit MQ 是建立在Erlang OTP平台上。所以在安裝rabbitMQ之前,需要先安裝Erlang 。
http://www.erlang.org/download/otp_win32_R16B03.exe,需要其他版本或者64位系統的,可以去官網下載。
全部點擊“下一步”就行。
有的選擇其他的安裝方式,可能需要添加一下系統環境變量
2.安裝RabbitMQ
下載運行 :http://www.rabbitmq.com/releases/rabbitmq-server/v3.2.3/rabbitmq-server-3.2.3.exe ,需要其他版本或者64位系統的,可以去官網下載。
依舊可以不改變默認進行安裝。
需要注意:默認安裝的RabbitMQ 監聽端口是5672
3.配置
激活 RabbitMQ's Management Plugin
使用RabbitMQ 管理插件,可以更好的可視化方式查看Rabbit MQ 服務器實例的狀態。
打開命令窗口:
輸入命令:
"C:\Program Files\RabbitMQ Server\rabbitmq_server-3.2.3\sbin\rabbitmq-plugins.bat" enable rabbitmq_management
重啟服務才行,使用命令:
net stop RabbitMQ && net start RabbitMQ
這時候,也許會出現這種結果:
“發生錯誤:發生系統錯誤 5。 拒絕訪問。”
問題解決方案:使用管理員打開cmd再執行此命令:
在C:\Windows\System32 找到cmd.exe 雙擊執行。
創建用戶,密碼,綁定角色
使用rabbitmqctl控制台命令(位於C:\Program Files\RabbitMQ Server\rabbitmq_server-xxx\sbin>)來創建用戶,密碼,綁定權限等。
注意:安裝路徑不同的請看仔細啊。
rabbitmq的用戶管理包括增加用戶,刪除用戶,查看用戶列表,修改用戶密碼。
新增一個用戶:
rabbitmqctl.bat add_user username password
查看已有用戶及用戶的角色:
rabbitmqctl.bat list_users
此時來看下我們當前用戶后面沒有“[administrator]”
這個administrator是干嘛用的呢?這就涉及到用戶角色問題了:
rabbitmq用戶角色可分為五類:超級管理員, 監控者, 策略制定者, 普通管理者以及其他。
(1) 超級管理員(administrator)
可登陸管理控制台(啟用management plugin的情況下),可查看所有的信息,並且可以對用戶,策略(policy)進行操作。
(2) 監控者(monitoring)
可登陸管理控制台(啟用management plugin的情況下),同時可以查看rabbitmq節點的相關信息(進程數,內存使用情況,磁盤使用情況等)
(3) 策略制定者(policymaker)
可登陸管理控制台(啟用management plugin的情況下), 同時可以對policy進行管理。
(4) 普通管理者(management)
僅可登陸管理控制台(啟用management plugin的情況下),無法看到節點信息,也無法對策略進行管理。
(5) 其他的
無法登陸管理控制台,通常就是普通的生產者和消費者
我們也給新建的用戶username 變成 “超級管理員” 角色:
rabbitmqctl.bat set_user_tags username administrator
修改密碼:
rabbitmqctl change_password userName newPassword
刪除已存在的用戶:
rabbitmqctl.bat delete_user username
這樣基本的配置就結束了。
然后我們打開瀏覽器 http://localhost:15672 訪問Rabbit Mq的管理控制台,使用剛才創建的賬號登陸系統:
設置虛擬主機和用戶關系(默認虛擬主機名為 "/”),選中admin菜單,然后選中右邊的 virtual hosts 菜單,如下圖:
在最下面有一個添加虛擬主機,我們自己起好名稱點擊 add virtual host 即可。
然后再上面的列表中就可以看到你新添加的主機,再點擊你新添加的虛擬主機,為其分配用戶,我們首先可以分配我們前面創建的username 用戶,然后也可以在admin→users頁面添加新的用戶后,再將其添加到當前新建的虛擬主機。
對於這塊虛擬主機和用戶,我個人的理解就像是你買了一輛車(用戶),我買了一輛車(用戶), 但是我們都得去交管所上拍照(虛擬主機)。
java代碼編寫
現在環境搭好,開始進行code。
新建一個maven工程,新手可參考 https://www.cnblogs.com/JJJ1990/p/8384386.html 前半部分。
在pom文件中添加jar包引用
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.6.5</version> </dependency>
發布者
1 public static void publisher() 2 { 3 Connection connection = null; 4 Channel channel = null; 5 try 6 { 7 ConnectionFactory factory = new ConnectionFactory(); 8 factory.setHost(ConnectionFactory.DEFAULT_HOST); //設置連接地址 默認 localhost 9 factory.setPort(ConnectionFactory.DEFAULT_AMQP_PORT); //監聽端口 默認5672 10 factory.setUsername("username"); //用戶名 11 factory.setPassword("password"); //密碼 12 factory.setVirtualHost("test_vhosts"); // 虛擬主機 13 //創建與RabbitMQ服務器的TCP連接 14 connection = factory.newConnection(); 15 channel = connection.createChannel(); //創建通道/頻道 16 channel.queueDeclare("JQueue", true, false, false, null); //參數中的 JQueue就是我們聲明的隊列名稱 17 String message = "First Message"; //消息體 18 channel.basicPublish("", "JQueue", null, message.getBytes()); 19 System.out.println("Send Message is:'" + message + "'"); 20 } 21 catch(Exception ex) 22 { 23 ex.printStackTrace(); 24 } 25 finally 26 { 27 if(channel != null) 28 { 29 try { 30 channel.close(); 31 } catch (IOException e) { 32 // TODO Auto-generated catch block 33 e.printStackTrace(); 34 } catch (TimeoutException e) { 35 // TODO Auto-generated catch block 36 e.printStackTrace(); 37 } 38 } 39 if(connection != null) 40 { 41 try { 42 connection.close(); 43 } catch (IOException e) { 44 // TODO Auto-generated catch block 45 e.printStackTrace(); 46 } 47 } 48 } 49 }
我們先新建一個publisher()方法,在引入jar包的時候要特別注意,同樣的名稱有好幾個包,我們統一都只引用 com.rabbitmq.client。
在queueDeclare方法中 除了隊列名稱外的其余4項參數意義如下
* @param durable true if we are declaring a durable queue (the queue will survive a server restart)
* @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
* @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
* @param arguments other properties (construction arguments) for the queue
消費者
1 public static void consumer() { 2 Connection connection = null; 3 Channel channel = null; 4 try 5 { 6 ConnectionFactory factory = new ConnectionFactory(); 7 factory.setHost(ConnectionFactory.DEFAULT_HOST); 8 factory.setPort(ConnectionFactory.DEFAULT_AMQP_PORT); 9 factory.setUsername("jjj"); //注意此處的用戶名和發布者不同,但是我將他們分配到了同一個虛擬主機,同樣可以獲取發布的消息 10 factory.setPassword("password"); 11 factory.setVirtualHost("test_vhosts"); 12 connection = factory.newConnection(); 13 channel = connection.createChannel(); 14 15 Consumer consumer = new DefaultConsumer(channel) { 16 @Override 17 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) 18 throws IOException { 19 String message = new String(body, "UTF-8"); 20 System.out.println(" Consumer have received '" + message + "'"); 21 } 22 }; 23 channel.basicConsume("JQueue", true, consumer); //綁定消費者和隊列 24 } 25 catch(Exception ex) 26 { 27 ex.printStackTrace(); 28 } 29 }
對比消費者和發布者的代碼我們可以發現,我用了不同的用戶連接同一個虛擬主機 test_vhosts
新建main方法
1 public static void main( String[] args ) 2 { 3 publisher(); 4 System.out.println( "Hello World!消息已經發布" ); 5 consumer(); 6 System.out.println( "Hello World!正在監聽接收" ); 7 }
運行代碼后可以看到如下結果:
瀏覽器打開rabbitmq界面。找到queues 菜單,點擊后,就可以在列表中找到我們建立的隊列,找到我剛才新建的jqueue隊列,點進去后,找到下面的 publish message 目錄
在Payload 中我們隨便輸入一些文字 點擊發布就可以看到代碼的控制台已經收到了這些消息
控制台輸出:
這說明程序運行后,消費者一直在監聽當前的隊列,一旦有消息,就會立刻獲取。