2.持久化、非持久化、租户、命名空间

目录

• 持久化的大白话说明
  • 持久化消息
  • 非持久化消息
• 持久&非持久消息发送方法
  • 生产者方法
  • 消费者方法
• 维护租户&命名空间
  • YML修改
  • 创建PulsarAdmin
  • 租户(增删改查)
  • 命名空间(增删查)
• 代码下载

持久化的大白话说明

Pulsar默认消息为持久化

持久化消息

只要消息产生了，除非所有硬盘都坏了，否则不会丢失

非持久化消息

消息存在内存中，服务挂了，或者消息产生式消费者未启动，都无法收到消息

持久&非持久消息发送方法

注：

• 显示指定持久|非持久都需要指定租户和命名空间
• pulsar默认命名租户为public，命名空间为default，会自动创建主题topic，但不会自动创建租户和命名空间
• 命名空间和租户不存在直接调用会报异常，解决办法为创建先创建租户和命名空间，再调用生产者和消费者，租户和命名空间操作方法见下方的《维护租户&命名空间》

生产者方法

    /**
     * 生产消息
     * @param msg
     * @param tag 0默认持久化的消息 1非持久化消息 2 持久化消息的另外一个写法，同0
     * @throws PulsarClientException
     */
    @GetMapping("/persistent/sendMsg")
    public MessageId sendMsg(String msg, String tag) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic="my-topic";
        if("1".equals(tag)) {
            topic="non-persistent://public/default/my-topic";
                   //非持久化         租户   命名空间   主题
        }
        if("2".equals(tag)){
            topic="persistent://public/default/my-topic";
                  //持久化         租户   命名空间   主题
        }
        Producer<byte[]> producer1 = pulsarFactory.newProducer()
                .topic(topic)
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：
        return producer1.send(msg.getBytes());
    }

消费者方法

注:监听方法同下

 /**
     * 手动执行获取消息
     * @param tag 0默认持久化的消息 1非持久化消息 2 持久化消息的另外一个写法，同0
     * @throws PulsarClientException
     */
    @GetMapping("/base/comsumer")
    public void comsumerByArtificial(String tag) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic="my-topic";
        if("1".equals(tag)) {
            topic="non-persistent://public/default/my-topic";
                   //非持久化         租户   命名空间   主题
        }
        if("2".equals(tag)){
            topic="persistent://public/default/my-topic";
                 //持久化         租户   命名空间   主题
        }
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscription")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }

维护租户&命名空间

操作下面的所有代码，需要另外引入Jar，可以引入pulsar-client和pulsar-client-admin，也可以只引入pulsar-client-all

如果通过RestApi或命令行来操作集群、租户、命名空间的操作，可以不看下面这些

        <dependency>
            <groupId>org.apache.pulsar</groupId>
            <artifactId>pulsar-client</artifactId>
            <version>2.8.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.pulsar</groupId>
            <artifactId>pulsar-client-admin</artifactId>
            <version>2.8.0</version>
        </dependency>
<!--同时引入上面两个或只引入下面一个-->
        <dependency>
            <groupId>org.apache.pulsar</groupId>
            <artifactId>pulsar-client-all</artifactId>
            <version>2.8.0</version>
        </dependency>

YML修改

新增serviceHttpUrl，此端口为apache-pulsar-2.8.0/conf/standalone.conf--->webServicePort=9090 配置，默认8081

server:
  port: 7777

pulsar:
  url: 10.30.54.103:6650
  serviceHttpUrl: http://10.30.54.103:9090
#  url: 10.30.54.103:6650,10.30.54.104:6650

创建PulsarAdmin

目的：用于操作租户和命名空间

• 修改PulsarConf.java

  新增serviceHttpUrl和pulsarAdminFactory

package com.project.pulsar.conf;

import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class PulsarConf {
    @Value("${pulsar.url}")
    String url;
    @Value("${pulsar.serviceHttpUrl}")
    String serviceHttpUrl;

    @Bean
    public PulsarClient pulsarFactory(){
        PulsarClient client = null;
        try {
            client = PulsarClient.builder()
                    .serviceUrl("pulsar://"+url)
                    .build();
        } catch (PulsarClientException e) {
        }
        return client;
    }
    @Bean
    public PulsarAdmin pulsarAdminFactory(){
        PulsarAdmin admin =null;
        try {
            admin = PulsarAdmin.builder()
                    .serviceHttpUrl(serviceHttpUrl)
                    .build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
        return  admin;
    }
}

租户(增删改查)

注：

• 此操作需要Pulsar集群名字，集群配置见官方文档
• 本配置为入门级，更多详细配置见此处

   /**
     * 创建租户
     * @throws Exception
     */
    @GetMapping("/persistent/createTenant")
    public void createTenant(String tenantName,String clusterName) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            Set set = new HashSet();
            set.add(clusterName);
            TenantInfo build = TenantInfo.builder().allowedClusters(set).build();
            tenants.createTenant(tenantName, build);
        }
    }
    /**
     * 获得现有租户
     * @throws Exception
     */
    @GetMapping("/persistent/tenantList")
    public List<String> tenantList() throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            return tenants.getTenants();
        }
        return null;
    }
    /**
     * 修改现有租户配置(所属集群等信息)
     * @throws Exception
     */
    @GetMapping("/persistent/updateTenant")
    public void updateTenant(String tenantName,String clusterName) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            Set set = new HashSet();
            set.add(clusterName);
            TenantInfo build = TenantInfo.builder().allowedClusters(set).build();
            tenants.updateTenant(tenantName, build);
        }
    }
    /**
     * 删除指定租户
     * @throws Exception
     */
    @GetMapping("/persistent/delTenant")
    public void delTenant(String tenantName) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            tenants.deleteTenant(tenantName);
        }
    }

命名空间(增删查)

改目前没有发现有什么意义，或许是没有需求吧，用到再说，都在官方文档中

注意创建命名空间时入参

/**
     * 创建命名空间
     * @param namespace 租户名字/命名空间 租户名字必须是已存在的
     * @throws Exception
     */
    @GetMapping("/persistent/createNamespace")
    public void createNamespace(String namespace) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Namespaces namespaces = admin.namespaces();
            namespaces.createNamespace(namespace);
        }
    }

    /**
     * 根据租户获得命名空间
     * @param tenant
     * @return
     * @throws Exception
     */
    @GetMapping("/persistent/namespaceList")
    public List<String> namespaceList(String tenant) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Namespaces namespaces = admin.namespaces();
            return namespaces.getNamespaces(tenant);
        }
        return null;
    }

    /**
     * 删除命名空间
     * @param npname
     * @throws Exception
     */
    @GetMapping("/persistent/delNamespace")
    public void delNamespace(String npname) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Namespaces namespaces = admin.namespaces();
            namespaces.deleteNamespace(npname);
        }
    }

代码下载

代码见此Persistent包下

更多AdminApi见此